feat(ai): Cursor subscription canary lane + provider scenario matrix

client/config.py

@@ -18,7 +18,7 @@
 class ProviderConfig:
     """Per-provider model configuration."""
 
-    label: str  # "claude" | "gpt" | "codex"
+    label: str  # "claude" | "gpt" | "codex" | "cursor"
     model: str
     can_execute_code: bool = False
     can_analyze: bool = True
@@ -35,6 +35,10 @@ def gpt(cls, model: str = "gpt-5.4-mini") -> "ProviderConfig":
     def codex(cls, model: str = "") -> "ProviderConfig":
         return cls(label="codex", model=model or "codex-cli", can_execute_code=True, can_analyze=True)
 
+    @classmethod
+    def cursor(cls, model: str = "") -> "ProviderConfig":
+        return cls(label="cursor", model=model or "cursor-agent", can_execute_code=True, can_analyze=True)
+
 
 @dataclass(frozen=True)
 class GatewayConfig:

client/gateway_client.py

@@ -25,7 +25,7 @@
 class AiResult:
     """Result from a single AI call."""
 
-    provider: str  # "claude" | "gpt" | "codex"
+    provider: str  # "claude" | "gpt" | "codex" | "cursor"
     model: str
     success: bool
     output: str = ""

docs/cursor-research-backend-2026-09-09.md

@@ -10,7 +10,7 @@
 
 Codex→Cursor 仅在请求显式允许、服务 `AI_GATEWAY_CURSOR_FALLBACK_ENABLED=true`、Codex 准入确认未开始且额度预留/账户能力不可用时选择。显式型号不跨后端换型。启动后超时、失败、结果缺失、通信不明不会触发另一后端；没有 API fallback。已有 `analyze/review` API consumer 保留原用途与预算保护。
 
-AAB 两个实际调用者 `run_research_task_diagnosis.py` 与 `run_portfolio_research_proposal_diagnosis.py` 通过 `AI_GATEWAY_RESEARCH_PROVIDERS=cursor` 或 `codex,cursor` 显式采用；默认 `codex`。其权限、去重、输出验收与 advisory 定位保持。SDK 安装和 QPK consumer 的采用必须单独验证，源代码可用不等于已安装。
+AAB 两个实际调用者 `run_research_task_diagnosis.py` 与 `run_portfolio_research_proposal_diagnosis.py` 通过 `AI_GATEWAY_RESEARCH_PROVIDERS=cursor` 或 `codex,cursor` 显式采用；默认 `codex`。其权限、去重、输出验收与 advisory 定位保持。场景命名与 canary 阶梯见 [provider-call-scenarios-2026-09-17.md](provider-call-scenarios-2026-09-17.md)。SDK 安装和 QPK consumer 的采用必须单独验证，源代码可用不等于已安装。
 
 ## 模型与费用
 
@@ -20,7 +20,7 @@ AAB 两个实际调用者 `run_research_task_diagnosis.py` 与 `run_portfolio_re
 
 Cursor 官方区分 Cursor Models 和 Other Models 两个消费池；第三方模型从 Other Models 按对应模型 API 单价消耗额度，超额可另外付费。型号可见不证明余额，也不证明免费。见 [Cursor 模型与价格](https://cursor.com/docs/models-and-pricing)。
 
-启用还要求 policy 的 `on_demand_disabled_verified=true`、未过期 `valid_until` 和全账户 `max_daily_calls`。默认示例未确认费用、已过期，不能执行。额度计数复用持久存储，跨 HTTP 请求串行准入与预留；计数区分 Cursor/Codex，Cursor 实际费用和余额保留未知，不能报为 0。损坏/缺配置的额度存储拒绝 Cursor；这些门不授予交易或候选晋级权限。
+启用还要求 policy 的 `on_demand_disabled_verified=true`（确认订阅不会溢出成付费 on-demand，不是 API 计费开关）、未过期 `valid_until` 和全账户 `max_daily_calls`。默认示例未确认、已过期，不能执行。额度计数复用持久存储，跨 HTTP 请求串行准入与预留；计数区分 Cursor/Codex，Cursor 实际费用和余额保留未知，不能报为 0，也不计入 API 美元预算。损坏/缺配置的额度存储拒绝 Cursor；这些门不授予交易或候选晋级权限。
 
 ## 目录刷新
 
@@ -32,7 +32,7 @@ Cursor 官方区分 Cursor Models 和 Other Models 两个消费池；第三方
 
 执行器每次新建临时任务工作区，复制服务拥有的 AGENTS 和研究真实性、故障验收两个 skill；这些文本与任务输入一起通过 stdin 提交。source_repository/source_ref 仅为来源元数据，不声称已 checkout 或读取对应代码。只接受 CLI 成功终态 `type=result/subtype=success/is_error=false` 的非空 result；失败固定脱敏。参见 [CLI 参数](https://cursor.com/docs/cli/reference/parameters)、[输出协议](https://cursor.com/docs/cli/reference/output-format)、[权限配置](https://cursor.com/docs/cli/reference/permissions)。
 
-当前固定 ask、sandbox enabled、禁自动更新；项目权限拒绝 Read/Shell/Write/MCP/WebFetch。额外传 `--allowed-tools "" --exclude-workspace-context`，并将规则与证据直接放入 stdin。AGENTS 和 skill 是任务指导；allowed-tools 是经 CLI 发送的服务端 no-tools 限制，不声称完整本地 OS 隔离。独立材料审查指出标准 Grep/Ls 不都消费 Read deny，原生 sandbox 默认 system read 不能描述成 workspace-only。根任务决定此最小首期可先部署 `AI_GATEWAY_CURSOR_ENABLED=false`，先验收无模型服务及目录刷新；自动启用仍须费用确认和实际 canary 的零工具调用验证。不通过不启用，不因沙箱失败改为 disabled。
+当前固定 ask、sandbox enabled、禁自动更新；项目权限拒绝 Read/Shell/Write/MCP/WebFetch。额外传 `--allowed-tools ""`（不传 `--exclude-workspace-context`：当前订阅/型号会 `invalid_argument`），并将规则与证据直接放入 stdin。服务若以 root 运行，用 `AI_GATEWAY_CURSOR_HOME` 指向已 `agent login` 的订阅 home。AGENTS 和 skill 是任务指导；allowed-tools 是经 CLI 发送的服务端 no-tools 限制，不声称完整本地 OS 隔离。独立材料审查指出标准 Grep/Ls 不都消费 Read deny，原生 sandbox 默认 system read 不能描述成 workspace-only。根任务决定此最小首期可先部署未确认费用的示例 policy 与目录刷新，不跑模型；真正执行仍须费用确认和实际 canary 的零工具调用验证。没有单独的 `AI_GATEWAY_CURSOR_ENABLED` 总开关——选型靠请求/`AI_GATEWAY_RESEARCH_PROVIDERS`，准入靠 policy 与 roster。不通过不启用，不因沙箱失败放宽门。
 
 ### 临时工作区的 headless 信任确认（2026-09-09）
 

docs/provider-call-scenarios-2026-09-17.md

@@ -0,0 +1,51 @@
+# Provider 调用场景与模式（2026-09-17）
+
+实现入口：`service/provider_scenarios.py`。消费者用 `resolve_execute_kwargs(...)` 取 execute 参数。
+
+## 三池经济模型
+
+| 池 | 载体 | 计费 | 准入 |
+|---|---|---|---|
+| Codex 订阅 | CLI execute | 订阅容量（非 API USD） | 账户 rate limits / 研究路由 |
+| Cursor 订阅 | CLI execute | 订阅容量；on-demand 须已确认关闭 | 可信 policy + roster + 日调用上限 |
+| OpenAI/Anthropic API | analyze / review | API 美元预算 | `api_budget_admission` |
+
+**API 永不顶替订阅执行。** 订阅 defer/额度满 → 停或延期，不改打 analyze/review。
+
+## Provider 选型（已定案）
+
+| 场景类 | Provider | 说明 |
+|---|---|---|
+| 诊断 / briefing（canary） | 默认 Codex；可显式 Cursor | 仅 `drift_analysis` / `research_summary` |
+| 晋级主审 / codegen / bugfix / 验收探针 | **Codex only** | 网关拒绝 Cursor |
+| dual-review secondary / analyze | **API** | 独立预算 |
+| Codex→Cursor fallback | **默认关** | 仅 canary stage + 显式链 + `AI_GATEWAY_CURSOR_FALLBACK_ENABLED`；启动后失败不换后端 |
+
+网关强制：`allowed_providers` 含 `cursor` 时，`research_stage` 必须属于 canary（`drift_analysis`/`research_summary`），否则 `cursor_stage_not_canary`。
+
+## 智能适配 vs 强制指定
+
+**智能适配**：场景填 `mode` / `stage` / `providers` / `complexity`；晋级 pin `xhigh`。型号留给订阅准入。
+
+**强制指定**：`allowed_providers` / `model` / `effort` / `complexity`；冲突 `ValueError`。固定 Codex 场景可覆盖 `research_stage`（仍不得选 Cursor）。
+
+## 启用阶梯
+
+1. 默认 Codex；Cursor 未确认 on-demand 关闭前不能跑。
+2. Canary：显式 `AI_GATEWAY_RESEARCH_PROVIDERS=cursor` + diagnosis（不要先开 fallback 链）。
+3. 再扩 portfolio → briefing。
+4. 晋级/codegen/bugfix 永不 Cursor；fallback 默认保持 false。
+
+## VPS 验收（2026-09-18）
+
+| 项 | 结果 |
+|---|---|
+| AppArmor + `cursor-sandbox-apparmor` + bubblewrap；`--sandbox enabled` | 可用（未改 disabled） |
+| `AI_GATEWAY_CURSOR_HOME` + 去掉 `--exclude-workspace-context` | 已部署（`31838f8`） |
+| `research_task_diagnosis` | PASS（`cursor-grok-4.6-medium`） |
+| `portfolio_proposal_diagnosis` | PASS |
+| `daily_briefing`（`research_summary`） | PASS（`cursor-grok-4.6-low`） |
+| Cursor @ `promotion_review` / `optimization` | `cursor_stage_not_canary` |
+| Fallback | 仍 false |
+
+HTTP `/v1/ai/execute/jobs` 仍需 OIDC（static token 拒执行）。日调用计数含上述 canary；policy `max_daily_calls` 耗尽前勿再压测。

ops/quant-monitor/scripts/health_cycle.py

@@ -396,13 +396,19 @@ def run_historical_diagnosis_rehearsal(
     except (ImportError, OSError, RuntimeError, ValueError):
         return {"status": "deferred", "reason": "ai_gateway_not_configured"}
     try:
+        from service.provider_scenarios import (
+            SCENARIO_RESEARCH_TASK_DIAGNOSIS,
+            resolve_execute_kwargs,
+        )
+
         result = client.execute(
             _historical_diagnosis_rehearsal_prompt(),
             task=_HISTORICAL_DIAGNOSIS_REHEARSAL_TASK,
-            mode="review_only",
+            **resolve_execute_kwargs(
+                SCENARIO_RESEARCH_TASK_DIAGNOSIS,
+                allowed_providers=["codex"],
+            ),
             sandbox="read-only",
-            research_stage="drift_analysis",
-            allowed_providers=["codex"],
             source_repository=_OPERATIONAL_DIAGNOSIS_SOURCE_REPOSITORY,
             source_ref="main",
             timeout=600,
@@ -462,13 +468,16 @@ def _run_operational_diagnosis(
     except OSError:
         return {"status": "deferred", "reason": "dedupe_state_unavailable"}
     try:
+        from service.provider_scenarios import (
+            SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS,
+            resolve_execute_kwargs,
+        )
+
         result = client.execute(
             _operational_diagnosis_prompt(data_errors, observation=observation),
             task="operational_data_diagnosis",
-            mode="review_only",
+            **resolve_execute_kwargs(SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS),
             sandbox="read-only",
-            research_stage="drift_analysis",
-            allowed_providers=["codex"],
             source_repository=_OPERATIONAL_DIAGNOSIS_SOURCE_REPOSITORY,
             source_ref="main",
             timeout=600,

scripts/codex_audit_service.py

@@ -1,9 +1,10 @@
 #!/usr/bin/env python3
-"""Codex audit service — authenticated VPS facade for Codex execution.
+"""Legacy Codex audit service entry — FROZEN compatibility artifact.
 
-The VPS service intentionally runs only Codex. Claude/GPT direct API fallbacks
-remain in caller-side GitHub workflows/scripts so provider API keys do not live
-in, or pass through, this service.
+Production VPS systemd runs ``python3 -m service.ai_gateway_service`` via
+``scripts/deploy_codex_audit_service.sh``. This module is retained only for
+offline historical/reference tests and must not be re-installed or started as
+the live service without a separate consumer audit.
 """
 
 from __future__ import annotations

scripts/deploy_codex_audit_service.sh

@@ -661,7 +661,8 @@ deploy() {
   local runner_user
   runner_user="$(id -un)"
 
-  install_file "scripts/codex_audit_service.py" "${DEPLOY_DIR}/scripts/codex_audit_service.py" "0755"
+  # Legacy scripts/codex_audit_service.py is frozen/compat-only and is no longer
+  # installed. Production systemd ExecStart uses service.ai_gateway_service.
   install_service_package
   sudo install -d -m 0700 -o "$runner_user" -g "$runner_user" "$JOB_DIR"
   write_default_execution_policy_if_missing

scripts/deploy_cursor_research.sh

@@ -31,7 +31,6 @@ if ! sudo test -e "$CURSOR_POLICY_ROOT/cursor_research.json"; then
 fi
 if ! sudo test -e "$CURSOR_CONFIG_ROOT/cursor.env"; then
   sudo tee "$CURSOR_CONFIG_ROOT/cursor.env" >/dev/null <<EOF
-AI_GATEWAY_CURSOR_ENABLED=false
 AI_GATEWAY_CURSOR_FALLBACK_ENABLED=false
 AI_GATEWAY_CURSOR_BIN=$CURSOR_EXECUTABLE
 AI_GATEWAY_CURSOR_POLICY_PATH=$CURSOR_POLICY_ROOT/cursor_research.json

scripts/research_summary.py

@@ -7,6 +7,8 @@
 from collections.abc import Callable, Mapping
 from typing import Any
 
+from service.provider_scenarios import SCENARIO_RESEARCH_SUMMARY, resolve_execute_kwargs
+
 
 def unavailable() -> dict[str, str]:
     return {"status": "unavailable", "text": "", "provider": "", "model": ""}
@@ -38,10 +40,16 @@ def summarize(summary_context):
                 f"\nDATA:\n{json.dumps(context, ensure_ascii=False, allow_nan=False, sort_keys=True)}"
             )
             result = client.execute(
-                prompt, mode="review_only", research_stage=research_stage,
-                allowed_providers=["codex"], source_repository=repository,
-                source_ref=revision, timeout=600,
+                prompt,
+                **resolve_execute_kwargs(
+                    SCENARIO_RESEARCH_SUMMARY,
+                    research_stage=research_stage,
+                ),
+                source_repository=repository,
+                source_ref=revision,
+                timeout=600,
             )
+
             if (result.success is not True or result.provider != "codex" or not result.model
                     or result.error or result.note or not isinstance(result.raw, dict)
                     or result.raw.get("status") != "succeeded"):

scripts/run_account_diagnosis.py

@@ -301,16 +301,18 @@ def run_diagnosis(
 
             client_factory = AiGatewayClient
         client = client_factory(config_loader())
+        from service.provider_scenarios import (
+            SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS,
+            resolve_execute_kwargs,
+        )
+
         result = client.execute(
             build_prompt(observation),
             task="account_operational_diagnosis",
-            mode="review_only",
+            **resolve_execute_kwargs(SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS),
             sandbox="read-only",
-            allowed_providers=["codex"],
             source_repository=SOURCE_REPOSITORY,
             source_ref=SOURCE_REF,
-            research_stage="drift_analysis",
-            complexity="high",
             timeout=600,
         )
     except Exception:

scripts/run_cn_index_etf_research.py

@@ -23,6 +23,7 @@
 from typing import Any
 from zoneinfo import ZoneInfo
 
+from service.provider_scenarios import SCENARIO_CN_INDEX_ETF_RESEARCH, resolve_execute_kwargs
 from service.research_task import validate_strategy_diagnosis_task
 
 PROFILE = "cn_index_etf_tactical_rotation"
@@ -444,8 +445,13 @@ def _diagnosis(runtime, drift, revision):
                               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)
+        result = client.execute(
+            runtime.prompt(context),
+            **resolve_execute_kwargs(SCENARIO_CN_INDEX_ETF_RESEARCH),
+            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")}

scripts/run_global_etf_research_codegen.py

@@ -489,6 +489,7 @@ def _validate_review(output: str) -> dict[str, str]:
 
 def _codex_execute(*, source_ref: str):
     from ai_gateway_client import AiGatewayClient, GatewayConfig
+    from service.provider_scenarios import SCENARIO_GLOBAL_ETF_CODEGEN, resolve_execute_kwargs
 
     config = GatewayConfig.from_env()
     if config.research_providers != ("codex",):
@@ -499,14 +500,14 @@ def execute(prompt: str):
         return client.execute(
             prompt,
             task=GLOBAL_ETF_RESEARCH_CODEGEN_TASK,
-            mode="review_only",
-            model=GLOBAL_ETF_RESEARCH_CODEGEN_MODEL,
-            complexity="medium",
-            research_stage="optimization",
+            **resolve_execute_kwargs(
+                SCENARIO_GLOBAL_ETF_CODEGEN,
+                model=GLOBAL_ETF_RESEARCH_CODEGEN_MODEL,
+                reasoning_effort="medium",
+                complexity="medium",
+            ),
             research_objective=GLOBAL_ETF_RESEARCH_OBJECTIVE,
-            reasoning_effort="medium",
             sandbox="read-only",
-            allowed_providers=["codex"],
             source_repository=GLOBAL_ETF_SOURCE_REPOSITORY,
             source_ref=source_ref,
             timeout=1800,

scripts/run_new_research.py

@@ -1826,6 +1826,7 @@ def _digest_map(raw: Any) -> dict[str, str]:
 def _codex_codegen_execute(*, source_ref: str, research_objective: str):
     """Return the bounded Codex-only patch callback for the SOXL lane."""
     from ai_gateway_client import AiGatewayClient, GatewayConfig
+    from service.provider_scenarios import SCENARIO_SOXL_RSI2_CODEGEN, resolve_execute_kwargs
 
     config = GatewayConfig.from_env()
     if config.research_providers != ("codex",):
@@ -1836,14 +1837,14 @@ def execute(prompt: str):
         return client.execute(
             prompt,
             task=SOXL_RSI2_CODEGEN_TASK,
-            mode="review_only",
-            model="gpt-5.6-luna",
-            complexity="medium",
-            research_stage="optimization",
+            **resolve_execute_kwargs(
+                SCENARIO_SOXL_RSI2_CODEGEN,
+                model="gpt-5.6-luna",
+                reasoning_effort="medium",
+                complexity="medium",
+            ),
             research_objective=research_objective,
-            reasoning_effort="medium",
             sandbox="read-only",
-            allowed_providers=["codex"],
             source_repository="QuantStrategyLab/AIAuditBridge",
             source_ref=source_ref,
             timeout=1800,
@@ -1859,6 +1860,7 @@ def _codex_callbacks(*, source_ref: str, facts: Mapping[str, Any]):
     advisory; lifecycle gates and source evidence remain deterministic.
     """
     from ai_gateway_client import AiGatewayClient, GatewayConfig
+    from service.provider_scenarios import SCENARIO_NEW_RESEARCH_DESIGN, resolve_execute_kwargs
 
     config = GatewayConfig.from_env()
     if config.research_providers != ("codex",):
@@ -1875,10 +1877,13 @@ def diagnose(context, _budget):
             f"\nLOCAL_FACTS:\n{encoded_facts}"
         )
         response = client.execute(
-            prompt, task="new_research_design", mode="review_only", complexity="low",
-            research_stage="optimization", sandbox="read-only",
-            allowed_providers=["codex"], source_repository="QuantStrategyLab/AIAuditBridge",
-            source_ref=source_ref, timeout=300,
+            prompt,
+            task="new_research_design",
+            **resolve_execute_kwargs(SCENARIO_NEW_RESEARCH_DESIGN),
+            sandbox="read-only",
+            source_repository="QuantStrategyLab/AIAuditBridge",
+            source_ref=source_ref,
+            timeout=300,
         )
         raw = response.raw if isinstance(response.raw, dict) else {}
         if not (response.success is True and raw.get("status") == "succeeded"

scripts/run_portfolio_research_proposal_diagnosis.py

@@ -34,6 +34,10 @@
     marker_for_portfolio_research_proposal,
     validate_portfolio_candidate_readiness,
 )
+from service.provider_scenarios import (  # noqa: E402
+    SCENARIO_PORTFOLIO_PROPOSAL_DIAGNOSIS,
+    resolve_execute_kwargs,
+)
 
 
 SOURCE_REPOSITORY = "QuantStrategyLab/UsEquitySnapshotPipelines"
@@ -185,9 +189,10 @@ def run_portfolio_research_proposal_diagnosis(
     try:
         ai_result = client.execute(
             prompt,
-            mode="review_only",
-            research_stage="drift_analysis",
-            **({"allowed_providers": list(config.research_providers)} if config.research_providers != ("codex",) else {}),
+            **resolve_execute_kwargs(
+                SCENARIO_PORTFOLIO_PROPOSAL_DIAGNOSIS,
+                research_providers=config.research_providers,
+            ),
             timeout=300,
             source_repository=repository,
         )