Pigbibi · Oct 7, 2026 · Oct 7, 2026
diff --git a/.github/actionlint.yaml b/.github/actionlint.yaml
@@ -0,0 +1,3 @@
+self-hosted-runner:
+  labels:
+    - research-runner
diff --git a/.github/workflows/engineering-material-review.yml b/.github/workflows/engineering-material-review.yml
@@ -0,0 +1,26 @@
+name: Frozen Engineering Material Review V2
+on:
+  workflow_dispatch:
+permissions:
+  contents: read
+  id-token: write
+concurrency:
+  group: engineering-material-review-v2
+  cancel-in-progress: false
+jobs:
+  review:
+    if: github.ref == 'refs/heads/main' && github.run_attempt == 1 && vars.AI_SERVICE_RELEASE_READY == 'true'
+    runs-on: [self-hosted, research-runner]
+    timeout-minutes: 35
+    env:
+      AI_SERVICE_URL: ${{ vars.AI_SERVICE_URL }}
+      AI_SERVICE_AUDIENCE: ${{ vars.AI_SERVICE_AUDIENCE }}
+      AI_SERVICE_REVIEWERS_JSON: ${{ vars.AI_SERVICE_REVIEWERS_JSON }}
+      AUDIT_MATERIAL: ${{ vars.AI_SERVICE_ENGINEERING_MATERIAL_PATH }}
+      AUDIT_ROLES: ${{ vars.AI_SERVICE_ENGINEERING_ROLES_JSON }}
+    steps:
+      - name: Review caller-owned frozen material using the approved runtime
+        run: |
+          set -euo pipefail
+          test -n "$AUDIT_MATERIAL" && test -n "$AUDIT_ROLES"
+          python -m quant_platform_kit.strategy_lifecycle.audit_tasks --material "$AUDIT_MATERIAL" --roles-json "$AUDIT_ROLES"
diff --git a/.github/workflows/reusable-drift-check.yml b/.github/workflows/reusable-drift-check.yml
@@ -24,10 +24,18 @@ on:
         required: false
         type: string
         default: "main"
-      ai_gateway_service_url:
+      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
@@ -71,8 +79,6 @@ on:
         type: string
         default: ""
     secrets:
-      codex_audit_service_url:
-        required: false
       snapshot_repository_token:
         required: false
 
@@ -83,6 +89,7 @@ permissions:
 
 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.
@@ -288,6 +295,25 @@ jobs:
           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: |
@@ -318,110 +344,9 @@ jobs:
           print({"created_issues": len(created), "results": len(results)})
           PY
 
-      - name: Checkout AIAuditBridge
-        uses: actions/checkout@v6
-        with:
-          repository: QuantStrategyLab/AIAuditBridge
-          ref: 9fe23596d3722777dfbd28eac1a710ab2b2d6431
-          path: external/AIAuditBridge
-
-      - name: Dual-review critical drift
+      - name: Review critical drift through task service
         env:
-          AIAUDIT_BRIDGE_ROOT: external/AIAuditBridge
-          CODEX_AUDIT_SERVICE_URL: ${{ secrets.codex_audit_service_url }}
-          AI_GATEWAY_SERVICE_URL: ${{ inputs.ai_gateway_service_url }}
-          GH_TOKEN: ${{ github.token }}
-        run: |
-          set -euo pipefail
-          emit_parked_record() {
-            python - "$1" "$2" <<'PY'
-          import hashlib
-          import json
-          import os
-          import sys
-
-          reason = sys.argv[1]
-          detail = sys.argv[2]
-          domain = os.environ["STRATEGY_DOMAIN"]
-          dedup_key = hashlib.sha256(f"drift-dual-review/{domain}/{reason}".encode()).hexdigest()
-          print(json.dumps({
-              "schema": "qsl.drift_dual_review_availability.v1",
-              "state": "PARKED",
-              "domain": domain,
-              "reason": reason,
-              "detail": detail[:500],
-              "dedup_key": dedup_key,
-              "next_action": "retry_on_next_drift_cycle",
-          }, sort_keys=True))
-          PY
-          }
-          script="external/AIAuditBridge/scripts/run_drift_dual_review.py"
-          if [ ! -f "$script" ]; then
-            emit_parked_record "review_script_unavailable" "$script"
-            echo "::notice::Dual review parked: the drift detector and issue sync remain active"
-            exit 0
-          fi
-          if [ -z "${CODEX_AUDIT_SERVICE_URL:-}" ]; then
-            emit_parked_record "codex_audit_service_unconfigured" "CODEX_AUDIT_SERVICE_URL is empty"
-            echo "::notice::Dual review parked: the drift detector and issue sync remain active"
-            exit 0
-          fi
-          review_output="${RUNNER_TEMP}/drift-dual-review.json"
-          set +e
-          PYTHONPATH=external/AIAuditBridge python "$script" \
-            --domain "${STRATEGY_DOMAIN}" --dispatch >"$review_output"
-          review_rc=$?
-          set -e
-          if [ ! -f "$review_output" ]; then
-            emit_parked_record "review_output_unavailable" "exit_code=$review_rc output_file_missing"
-            echo "::warning::Dual review parked; no review output was produced"
-            exit 0
-          fi
-          cat "$review_output"
-          if [ "$review_rc" -ne 0 ]; then
-            review_state="$(python - "$review_output" <<'PY'
-          import json
-          import sys
-
-          try:
-              payload = json.load(open(sys.argv[1], encoding="utf-8"))
-          except (OSError, json.JSONDecodeError) as exc:
-              print(f"invalid_review_json:{exc}")
-              raise SystemExit(0)
-          if not isinstance(payload, dict):
-              print("invalid_review_json:top_level_not_object")
-              raise SystemExit(0)
-          if payload.get("degraded") is True:
-              print("provider_degraded")
-              raise SystemExit(0)
-          results = payload.get("results")
-          completed_outcomes = {"fail", "disagreement"}
-          if (
-              isinstance(results, list)
-              and results
-              and all(isinstance(result, dict) for result in results)
-              and all(str(result.get("outcome") or "") in completed_outcomes for result in results)
-          ):
-              print("review_completed_blocked")
-              raise SystemExit(0)
-          print("review_process_failed")
-          PY
-            )"
-            if [ "$review_state" = "review_completed_blocked" ]; then
-              emit_parked_record "review_completed_blocked" "exit_code=$review_rc completed safety review blocked promotion"
-              echo "::notice::Dual review completed and blocked promotion; drift remains parked"
-              exit 0
-            fi
-            if [ "$review_state" = "provider_degraded" ]; then
-              emit_parked_record "review_provider_degraded" "exit_code=$review_rc $review_state"
-              echo "::warning::Dual review unavailable/degraded; fail-closed without treating as completed veto"
-              exit 3
-            fi
-            review_reason="review_process_failed"
-            if [[ "$review_state" == invalid_review_json:* ]]; then
-              review_reason="review_output_invalid_json"
-            fi
-            emit_parked_record "$review_reason" "exit_code=$review_rc $review_state"
-            echo "::warning::Dual review parked; the drift detector and de-duplicated GitHub issue sync remain active"
-            exit 0
-          fi
+          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/README.md b/README.md
@@ -130,3 +130,7 @@ See [LICENSE](LICENSE).
 ## v1 migration
 
 `quant_platform_kit.strategy_contracts` was removed in v1. Import strategy contracts from `quant_platform_kit.common.strategy_contracts`, execution translation from `quant_platform_kit.common.execution_translation`, and runtime inputs from `quant_platform_kit.common.runtime_inputs`. No compatibility facade is provided.
+
+### AI 任务服务升级
+
+生命周期 AI 调用的本地 2.0 迁移使用通用任务接口，区分 API 与常驻助手。新路由配置、批准客户端产物、原任务恢复及研究权限边界见 [迁移说明](docs/ai-service/TASK-SERVICE-V2.zh-CN.md)。本次版本尚未发布或部署。
diff --git a/docs/ai-service/TASK-SERVICE-V2.zh-CN.md b/docs/ai-service/TASK-SERVICE-V2.zh-CN.md
@@ -0,0 +1,42 @@
+# 生命周期 AI 任务接口迁移
+
+本地升级基线：QuantPlatformKit `350dd38ece0952beef096893a7dcf8ae3871609a`。尚未提交、发布或部署；调用方的现有远端依赖固定版本不会自动包含这些改动。
+
+`strategy_lifecycle/ai_provider.py` 现在使用 `ai_service.client.TaskClient`，只调用 V2 的提交和查询接口。已经删除该模块的旧 HTTP 客户端、CLI/VPS 路由、内置模型、provider chain 和付费 fallback。`AiProviderConfig` 接收 `label`、`mode`、`model`、`profile`；不再提供旧的 `codex_vps()`、`claude()`、`gpt()` 工厂。
+
+依赖由批准环境安装本次 PersonalAIService 2.0 源码构建的 wheel。这个包尚未公开发布，不能从公共索引安装同名包代替。QuantPlatformKit 的基础功能不会主动加载 AI 客户端；开启 AI 路径前需要准备该 wheel 和 `AI_SERVICE_URL`、`AI_SERVICE_AUDIENCE`，身份使用批准的 token 客户端或 GitHub Actions OIDC。旧的 `CODEX_AUDIT_SERVICE_*` 和 `AI_GATEWAY_RESEARCH_PROVIDERS` 不再用于本模块。
+
+## 路由与结果
+
+普通执行读取 `AI_SERVICE_MODE`（默认 agent）、`AI_SERVICE_MODEL`（必须显式设置）、`AI_SERVICE_PROFILE`（默认 default）。审查从 `AI_SERVICE_REVIEWERS_JSON` 读取明确的路由列表；可选的核对路径由 `AI_SERVICE_VERIFIER_JSON` 指定，默认没有核对助手。例如：
+
+```json
+[
+  {"label": "reviewer-primary", "mode": "agent", "model": "configured-dot", "profile": "default"},
+  {"label": "reviewer-secondary", "mode": "agent", "model": "configured-grok", "profile": "second-opinion"}
+]
+```
+
+服务端决定实际 provider 和项目权限。这里的 label 是审查角色，不证明模型厂商或实际模型。同一 mode/profile 不能作为两个不同审查路径，但不同 profile 仍不证明云电脑、账户或工具权限隔离；正式独立审查需要产品端完成独立连接及权限隔离。
+
+任务请求只含通用目标、材料、输出 schema、超时、模式和 profile。生命周期输出被封装为一个 `report` 字符串，里面的业务 JSON 仍由生命周期调用方验证。读取结果必须匹配任务 ID 和原始请求，完成且标记 advisory 后才进入业务解析。API 路由核对接口报告的模型；常驻助手必须明确标记 `model_verification=unavailable`，不伪造旧 Codex 的模型或 reasoning-effort 证明。
+
+## 重试与研究权限
+
+调用方提供操作幂等键，或在 Actions 中使用同一 run ID；角色、材料、模型和路由参与摘要。生命周期研究决定使用冻结输入构造的完整 prompt 摘要作为操作身份。客户端超时及已知任务读取失败保留任务 ID，结果保持 outcome_unknown，没有自动回退或重新触发助手。
+
+研究流程将 ai_task_pending 保存为 deferred，不把它当作负面研究建议。后续通过专门的 pending reader 查询原任务；`resume_task_id` 路径不会提交新任务，并重新核对原始请求绑定。如果没有 reader，维持等待；配置或材料不一致时不能采用原结果。既有历史 Codex 容量延后记录的解析保留，以免改写既有研究证据；这不提供旧服务接口兼容。
+
+AI 的完成与推荐不授予数值研究、交易、资金或发布权限。原有来源新鲜度、冻结输入、Python 数值检验、回测、paired shadow、风险条件及人工候选决策继续生效。历史 `codex_integration.py` 文件名及既有证据记录没有机械改名。
+
+## 本地验收
+
+测试只使用 synthetic 数据与假任务客户端，未调用真实模型或读取真实市场、账户数据。当前已覆盖任务结果绑定、API 模型不匹配、常驻模型不可验证、幂等键、未知结果、超时、原任务恢复、双审查角色、研究恢复及风险准入。离线验证不代表 Dot/Grok 连接或生产研究已恢复。
+
+## 通用材料与审查
+
+`ai_patch.py` 只允许调用方显式允许的现有文件、基准摘要匹配且唯一定位的有界替换；业务语法或领域校验由调用方提供，全部预检通过才写候选。共享实现没有平台或策略路径白名单。
+
+`research_task.py` 保留已有 watcher wire schema，校验完整来源和严格类型的研究权限；不含 SOXL 的固定参数。非空参数边界摘要需要调用方明确提供允许值。策略仓库再检查其业务来源是否有资格使用该边界。
+
+`research_summary.py` 的短解释默认 API 模式，模型和 profile 明确配置；没有配置时返回 unavailable。`task_review.py` 接收调用方要求的审查角色，校验任务身份、角色唯一、任务唯一及完整意见格式；所有角色完成才能形成 advisory quorum，意见不一致需人工判断。它不发送外部消息，也不批准交易、部署或发布。角色和不同 profile 仍不证明产品端账号与工具隔离，正式采用前需要核对真实连接边界。
diff --git a/src/quant_platform_kit/strategy_lifecycle/ai_patch.py b/src/quant_platform_kit/strategy_lifecycle/ai_patch.py
@@ -0,0 +1,110 @@
+"""Bounded candidate edits; the caller supplies exact paths and domain validation."""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from pathlib import Path, PurePosixPath
+
+
+class PatchError(ValueError):
+    pass
+
+
+def parse_patch(text: str, *, max_changes: int = 20, max_edits: int = 20,
+                max_replacement_bytes: int = 128 * 1024):
+    if not isinstance(text, str):
+        raise PatchError("patch must be JSON text")
+    def pairs(items):
+        result = {}
+        for key, value in items:
+            if key in result:
+                raise PatchError("duplicate patch key")
+            result[key] = value
+        return result
+    try:
+        payload = json.loads(text, object_pairs_hook=pairs,
+                             parse_constant=lambda _: (_ for _ in ()).throw(PatchError("nonfinite patch value")))
+        if not isinstance(payload, dict) or set(payload) != {"final_message", "changes"}:
+            raise PatchError("invalid patch fields")
+        if not isinstance(payload["final_message"], str) or not isinstance(payload["changes"], list):
+            raise PatchError("invalid patch values")
+        if len(payload["changes"]) > max_changes:
+            raise PatchError("too many changed files")
+        replacement_bytes = 0
+        paths = set()
+        for change in payload["changes"]:
+            if not isinstance(change, dict) or set(change) != {"path", "base_sha256", "edits"}:
+                raise PatchError("targeted edits required")
+            path = change["path"]
+            if (not isinstance(path, str) or not path or "\\" in path
+                    or PurePosixPath(path).is_absolute() or PurePosixPath(path).as_posix() != path
+                    or any(p in {".", "..", ".git"} for p in PurePosixPath(path).parts)):
+                raise PatchError("invalid patch path")
+            if path in paths:
+                raise PatchError("duplicate changed file")
+            paths.add(path)
+            if not isinstance(change["base_sha256"], str) or not re.fullmatch(r"[0-9a-f]{64}", change["base_sha256"]):
+                raise PatchError("invalid base digest")
+            edits = change["edits"]
+            if not isinstance(edits, list) or not 1 <= len(edits) <= max_edits:
+                raise PatchError("invalid edit count")
+            for edit in edits:
+                if (not isinstance(edit, dict) or set(edit) != {"old", "new"}
+                        or not isinstance(edit["old"], str) or not edit["old"]
+                        or not isinstance(edit["new"], str) or edit["old"] == edit["new"]):
+                    raise PatchError("invalid targeted edit")
+                edit["old"].encode("utf-8")
+                replacement_bytes += len(edit["new"].encode("utf-8"))
+            if replacement_bytes > max_replacement_bytes:
+                raise PatchError("replacement limit exceeded")
+        return payload["final_message"].strip(), payload["changes"]
+    except (TypeError, UnicodeError, json.JSONDecodeError):
+        raise PatchError("invalid patch JSON") from None
+
+
+def apply_patch(root: Path, changes: list[dict], *, allowed_paths: frozenset[str], validate_updated=None):
+    # Revalidate callers' direct dictionaries, including all files before any write.
+    _, changes = parse_patch(json.dumps({"final_message": "", "changes": changes}, allow_nan=False))
+    root = Path(root)
+    if root.is_symlink() or not root.is_dir():
+        raise PatchError("candidate root must be a directory")
+    root = root.resolve()
+    prepared = []
+    for change in changes:
+        path = change["path"]
+        if path not in allowed_paths:
+            raise PatchError("path outside caller allowlist")
+        target = root / path
+        for part in (target, *target.parents):
+            if part == root:
+                break
+            if part.is_symlink():
+                raise PatchError("symlink in candidate path")
+        if not target.is_file():
+            raise PatchError("candidate file is missing")
+        original = target.read_bytes()
+        if hashlib.sha256(original).hexdigest() != change["base_sha256"]:
+            raise PatchError("candidate base digest mismatch")
+        try:
+            source = original.decode("utf-8")
+        except UnicodeError:
+            raise PatchError("candidate must be UTF-8") from None
+        locations = []
+        for edit in change["edits"]:
+            first = source.find(edit["old"])
+            if first < 0 or source.find(edit["old"], first + 1) >= 0:
+                raise PatchError("edit source must occur exactly once")
+            locations.append((first, first + len(edit["old"]), edit["new"]))
+        locations.sort()
+        if any(right[0] < left[1] for left, right in zip(locations, locations[1:])):
+            raise PatchError("overlapping candidate edits")
+        updated = source
+        for start, end, replacement in reversed(locations):
+            updated = updated[:start] + replacement + updated[end:]
+        if validate_updated is not None:
+            validate_updated(path, source, updated)
+        prepared.append((target, updated.encode("utf-8")))
+    for target, content in prepared:
+        target.write_bytes(content)
+    return [change["path"] for change in changes]