diff --git a/client/config.py b/client/config.py index a5205ff4..c5057658 100644 --- a/client/config.py +++ b/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: diff --git a/client/gateway_client.py b/client/gateway_client.py index c523b314..a1fe8f0b 100644 --- a/client/gateway_client.py +++ b/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 = "" diff --git a/docs/cursor-research-backend-2026-09-09.md b/docs/cursor-research-backend-2026-09-09.md index dddb4723..72c59cd8 100644 --- a/docs/cursor-research-backend-2026-09-09.md +++ b/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） diff --git a/docs/provider-call-scenarios-2026-09-17.md b/docs/provider-call-scenarios-2026-09-17.md new file mode 100644 index 00000000..f5c22a15 --- /dev/null +++ b/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` 耗尽前勿再压测。 \ No newline at end of file diff --git a/ops/quant-monitor/scripts/health_cycle.py b/ops/quant-monitor/scripts/health_cycle.py index 09e554c1..1202523d 100755 --- a/ops/quant-monitor/scripts/health_cycle.py +++ b/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, diff --git a/scripts/codex_audit_service.py b/scripts/codex_audit_service.py index cd1ac1ca..ec366c43 100644 --- a/scripts/codex_audit_service.py +++ b/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 diff --git a/scripts/deploy_codex_audit_service.sh b/scripts/deploy_codex_audit_service.sh index bee1dd56..4a26fafe 100644 --- a/scripts/deploy_codex_audit_service.sh +++ b/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 diff --git a/scripts/deploy_cursor_research.sh b/scripts/deploy_cursor_research.sh index db467659..cb280157 100755 --- a/scripts/deploy_cursor_research.sh +++ b/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 < 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"): diff --git a/scripts/run_account_diagnosis.py b/scripts/run_account_diagnosis.py index e22d5538..8d425422 100644 --- a/scripts/run_account_diagnosis.py +++ b/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: diff --git a/scripts/run_cn_index_etf_research.py b/scripts/run_cn_index_etf_research.py index 43413ba0..11aaa156 100644 --- a/scripts/run_cn_index_etf_research.py +++ b/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")} diff --git a/scripts/run_global_etf_research_codegen.py b/scripts/run_global_etf_research_codegen.py index e1be15a7..64af4965 100644 --- a/scripts/run_global_etf_research_codegen.py +++ b/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, diff --git a/scripts/run_new_research.py b/scripts/run_new_research.py index 90a8b185..4f632251 100644 --- a/scripts/run_new_research.py +++ b/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" diff --git a/scripts/run_portfolio_research_proposal_diagnosis.py b/scripts/run_portfolio_research_proposal_diagnosis.py index b9798bd2..44d1e99b 100644 --- a/scripts/run_portfolio_research_proposal_diagnosis.py +++ b/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, ) diff --git a/scripts/run_research_task_diagnosis.py b/scripts/run_research_task_diagnosis.py index ab060872..e9f1282c 100644 --- a/scripts/run_research_task_diagnosis.py +++ b/scripts/run_research_task_diagnosis.py @@ -30,6 +30,10 @@ from client.config import GatewayConfig # noqa: E402 from client.gateway_client import AiGatewayClient # noqa: E402 +from service.provider_scenarios import ( # noqa: E402 + SCENARIO_RESEARCH_TASK_DIAGNOSIS, + resolve_execute_kwargs, +) from service.research_diagnosis import ( # noqa: E402 build_research_diagnosis_prompt, build_research_diagnosis_request, @@ -473,9 +477,10 @@ def run_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_RESEARCH_TASK_DIAGNOSIS, + research_providers=config.research_providers, + ), timeout=600, source_repository=str(request["target"]["repository"]), source_ref=str(request["target"]["strategy_revision"]), diff --git a/scripts/run_russell_research_explanation.py b/scripts/run_russell_research_explanation.py index 907e9fba..ec781f08 100644 --- a/scripts/run_russell_research_explanation.py +++ b/scripts/run_russell_research_explanation.py @@ -140,11 +140,20 @@ def explain(*, log_path: Path, run_path: Path, output_path: Path, source_ref: st config = GatewayConfig.from_env() if config.research_providers != ("codex",): raise RussellInputError("Codex-only research route is not configured") + from service.provider_scenarios import SCENARIO_RESEARCH_SUMMARY, resolve_execute_kwargs + response = AiGatewayClient(config).execute( - build_prompt(input_record), task="russell_research_explanation", mode="review_only", - complexity="low", research_stage="research_summary", reasoning_effort="low", - sandbox="read-only", allowed_providers=["codex"], source_repository="QuantStrategyLab/AIAuditBridge", - source_ref=source_ref, timeout=300, + build_prompt(input_record), + task="russell_research_explanation", + **resolve_execute_kwargs( + SCENARIO_RESEARCH_SUMMARY, + complexity="low", + reasoning_effort="low", + ), + 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 response.provider == "codex" and response.output and raw.get("status") == "succeeded" and raw.get("provider") == "codex" and raw.get("research_stage") == "research_summary"): diff --git a/scripts/run_semantic_quality_acceptance.py b/scripts/run_semantic_quality_acceptance.py index 6345c7c8..e7dbe949 100644 --- a/scripts/run_semantic_quality_acceptance.py +++ b/scripts/run_semantic_quality_acceptance.py @@ -10,6 +10,10 @@ from client.config import GatewayConfig from client.gateway_client import AiGatewayClient +from service.provider_scenarios import ( + SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE, + resolve_execute_kwargs, +) from service.research_diagnosis import build_research_diagnosis_prompt, build_research_diagnosis_request from tests.test_research_diagnosis import _semantic_quality_triggers, _task @@ -69,12 +73,12 @@ def run_acceptance( result = client.execute( build_research_diagnosis_prompt(request), task="execute", - mode="review_only", - model=MODEL, - research_stage="drift_analysis", - reasoning_effort=REASONING_EFFORT, + **resolve_execute_kwargs( + SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE, + model=MODEL, + reasoning_effort=REASONING_EFFORT, + ), sandbox="read-only", - allowed_providers=["codex"], source_repository=str(request["target"]["repository"]), source_ref=str(request["target"]["strategy_revision"]), timeout=600, diff --git a/scripts/run_soxl_manual_learning.py b/scripts/run_soxl_manual_learning.py index 36a2ade9..68d82e3b 100644 --- a/scripts/run_soxl_manual_learning.py +++ b/scripts/run_soxl_manual_learning.py @@ -20,6 +20,7 @@ from client.config import GatewayConfig from client.gateway_client import AiGatewayClient +from service.provider_scenarios import SCENARIO_SOXL_MANUAL_LEARNING, resolve_execute_kwargs from service.research_diagnosis import ( build_research_diagnosis_request, marker_for_research_diagnosis, @@ -1251,9 +1252,11 @@ def run_manual_learning( try: config = gateway_config or GatewayConfig.from_env() result = client_factory(config).execute( - _prompt(context, values, manifest_sha256), mode="review_only", - research_stage="optimization", allowed_providers=["codex"], - source_repository=EXPECTED_REPOSITORY, source_ref="main", timeout=600, + _prompt(context, values, manifest_sha256), + **resolve_execute_kwargs(SCENARIO_SOXL_MANUAL_LEARNING), + source_repository=EXPECTED_REPOSITORY, + source_ref="main", + timeout=600, ) except Exception: # noqa: BLE001 - provider detail must not cross this boundary artifact["failure_stage"] = "codex_unavailable" diff --git a/scripts/run_watchdog_repair_rehearsal.py b/scripts/run_watchdog_repair_rehearsal.py index 919e5602..ac504e35 100644 --- a/scripts/run_watchdog_repair_rehearsal.py +++ b/scripts/run_watchdog_repair_rehearsal.py @@ -286,13 +286,19 @@ def run_rehearsal( except Exception: return _park("AI_GATEWAY_NOT_CONFIGURED") try: + from service.provider_scenarios import ( + SCENARIO_RESEARCH_TASK_DIAGNOSIS, + resolve_execute_kwargs, + ) + result = client.execute( _prompt(), task="historical_watchdog_repair_rehearsal", - 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=SOURCE_REPOSITORY, source_ref="main", timeout=600, diff --git a/service/adapters/__init__.py b/service/adapters/__init__.py index d5a38600..92a4d511 100644 --- a/service/adapters/__init__.py +++ b/service/adapters/__init__.py @@ -1,5 +1,7 @@ """AiGateway adapters — pluggable AI backend implementations.""" -from service.adapters.llm_adapter import LlmAdapter from service.adapters.codex_adapter import CodexAdapter +from service.adapters.cursor_adapter import CursorAdapter +from service.adapters.execution import resolve_execution_adapter +from service.adapters.llm_adapter import LlmAdapter -__all__ = ["LlmAdapter", "CodexAdapter"] +__all__ = ["CodexAdapter", "CursorAdapter", "LlmAdapter", "resolve_execution_adapter"] diff --git a/service/adapters/cursor_adapter.py b/service/adapters/cursor_adapter.py index a4a0318d..22f120f6 100644 --- a/service/adapters/cursor_adapter.py +++ b/service/adapters/cursor_adapter.py @@ -31,6 +31,12 @@ def execute(self, *, prompt, sandbox='read-only', model=None, reasoning_effort=N 'allow': [], 'deny': _DENY, }})) env = {k: v for k, v in _codex_env().items() if not k.startswith(('AI_GATEWAY_', 'CURSOR_'))} + # Service may run as root; Cursor CLI auth lives under the + # subscription login home. Isolation stays disposable workspace + # + deny-all tools + sandbox enabled (not OS-user separation). + cursor_home = os.environ.get('AI_GATEWAY_CURSOR_HOME', '').strip() + if cursor_home: + env['HOME'] = cursor_home instructions = '\n\n'.join(path.read_text() for path in ( workspace / 'AGENTS.md', workspace / '.agents/skills/research-evidence/SKILL.md', @@ -38,9 +44,11 @@ def execute(self, *, prompt, sandbox='read-only', model=None, reasoning_effort=N )) # Trust only this service-created disposable directory, never # a caller checkout; tool permissions and sandbox stay separate. + # Do not pass --exclude-workspace-context: current Cursor + # subscription/models reject it (invalid_argument). completed = subprocess.run([ executable, '--disable-auto-update', '--print', '--output-format', 'json', '--mode', 'ask', - '--allowed-tools', '', '--exclude-workspace-context', + '--allowed-tools', '', '--sandbox', 'enabled', '--workspace', directory, '--trust', '--model', model, ], input=instructions + '\n\nTask and supplied evidence:\n' + prompt, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, diff --git a/service/adapters/execution.py b/service/adapters/execution.py new file mode 100644 index 00000000..92e93e37 --- /dev/null +++ b/service/adapters/execution.py @@ -0,0 +1,24 @@ +"""Thin execution-adapter selection for Codex / Cursor CLI backends.""" + +from __future__ import annotations + +from service.adapters.codex_adapter import CodexAdapter +from service.adapters.cursor_adapter import CursorAdapter +from service.contracts import PROVIDER_CODEX, PROVIDER_CURSOR + + +def resolve_execution_adapter(provider: str): + """Return the CLI execution adapter for an admitted provider. + + Unknown providers fail closed. Empty provider defaults to Codex for + backward-compatible sync/async execute paths that omit the field. + """ + selected = str(provider or PROVIDER_CODEX).strip().lower() or PROVIDER_CODEX + if selected == PROVIDER_CURSOR: + return CursorAdapter() + if selected == PROVIDER_CODEX: + return CodexAdapter() + raise ValueError(f"unsupported execution provider: {provider!r}") + + +__all__ = ["resolve_execution_adapter"] diff --git a/service/ai_gateway_service.py b/service/ai_gateway_service.py index 41308c6e..8cb27366 100644 --- a/service/ai_gateway_service.py +++ b/service/ai_gateway_service.py @@ -1,12 +1,12 @@ #!/usr/bin/env python3 """AiGateway — unified HTTP service for QuantStrategyLab AI calls. -Three endpoints, two adapters, one service. +Endpoints with API + CLI execution adapters (Codex / dormant Cursor lane). Hardened with rate limiting, input validation, audit logging, and sandbox controls. Endpoints: POST /v1/ai/analyze sync — LlmAdapter (Claude/GPT API) - POST /v1/ai/execute/jobs async — CodexAdapter (codex exec), poll via GET + POST /v1/ai/execute/jobs async — CodexAdapter or CursorAdapter (CLI), poll via GET POST /v1/ai/review sync — LlmAdapter × N + optional CodexAdapter Backward-compatible aliases: @@ -47,8 +47,8 @@ parse_review_request, ) from service.adapters.llm_adapter import DEFAULT_MAX_TOKENS, LlmAdapter, resolve_model -from service.adapters.cursor_adapter import CursorAdapter from service.adapters.codex_adapter import CodexAdapter +from service.adapters.execution import resolve_execution_adapter from service.model_resolver import resolve_codex_research_route from service.ai_provenance import ( build_provenance_receipt, @@ -167,6 +167,7 @@ def try_record_platform_execution(*_args, **_kwargs): _RATE_LIMIT_WINDOW_SECONDS = 60 _RATE_LIMIT_MAX_REQUESTS = 30 _analyze_timestamps: list[float] = [] +_RATE_LIMIT_LOCK = threading.Lock() SERVICE_FAILURE_CATEGORY_PATTERN = re.compile(r"\[([a-z_]+_failure)\]") @@ -299,13 +300,19 @@ def _resolve_codex_reasoning_effort(payload: dict[str, Any], task: str) -> str: def _admit_codex_execute(quota: Any, repo: str, payload: dict[str, Any]) -> dict[str, Any] | None: """Choose a Codex research route before consuming quota or starting a job.""" + from service.provider_scenarios import cursor_canary_research_stages + providers = payload.get("allowed_providers", ["codex"]) _validate_platform_bugfix_payload(payload) _validate_soxl_rsi2_codegen_payload(payload) _validate_global_etf_research_codegen_payload(payload) payload["provider"] = "codex" - if "cursor" in providers and not payload.get("research_stage"): - return {"status": "deferred", "error": "cursor_research_stage_required", "retry_at": None, "execution_started": False} + stage = str(payload.get("research_stage") or "").strip() + if "cursor" in providers: + if not stage: + return {"status": "deferred", "error": "cursor_research_stage_required", "retry_at": None, "execution_started": False} + if stage not in cursor_canary_research_stages(): + return {"status": "deferred", "error": "cursor_stage_not_canary", "retry_at": None, "execution_started": False} if providers == ["cursor"]: return _admit_cursor_execute(quota, repo, payload) if "research_stage" in payload: @@ -323,6 +330,7 @@ def _admit_codex_execute(quota: Any, repo: str, payload: dict[str, Any]) -> dict if route["action"] != "run": if (providers == ["codex", "cursor"] and os.environ.get("AI_GATEWAY_CURSOR_FALLBACK_ENABLED", "").lower() == "true" + and stage in cursor_canary_research_stages() and route["reason"] in {"codex_quota_reserved", "codex_account_unavailable", "codex_account_stale"} and str(payload.get("model") or "") in {"", "auto"}): return _admit_cursor_execute(quota, repo, payload) @@ -500,6 +508,11 @@ def _platform_bugfix_manual_mode(payload: dict[str, Any]) -> bool: def _admit_cursor_execute(quota: Any, repo: str, payload: dict[str, Any]) -> dict[str, Any] | None: from service.cursor_account import cursor_research_route + from service.provider_scenarios import cursor_canary_research_stages + + stage = str(payload.get("research_stage") or "").strip() + if stage not in cursor_canary_research_stages(): + return {"status": "deferred", "error": "cursor_stage_not_canary", "retry_at": None, "execution_started": False} cursor_payload = dict(payload) levels = ["low", "medium", "high"] cursor_payload["complexity"] = max((_normalize_complexity(str(payload.get("complexity") or "")) or "low", _estimate_codex_complexity(payload)), key=levels.index) @@ -519,13 +532,42 @@ def _admit_cursor_execute(quota: Any, repo: str, payload: dict[str, Any]) -> dic def _check_rate_limit(max_per_window: int = _RATE_LIMIT_MAX_REQUESTS, window: float = _RATE_LIMIT_WINDOW_SECONDS) -> None: """Sliding-window rate limiter for sync endpoints (analyze, review).""" global _analyze_timestamps - now = time.time() - _analyze_timestamps = [t for t in _analyze_timestamps if now - t < window] - if len(_analyze_timestamps) >= max_per_window: - raise PermissionError( - f"rate limit exceeded: {max_per_window} requests per {window:.0f}s" + with _RATE_LIMIT_LOCK: + now = time.time() + _analyze_timestamps = [t for t in _analyze_timestamps if now - t < window] + if len(_analyze_timestamps) >= max_per_window: + raise PermissionError( + f"rate limit exceeded: {max_per_window} requests per {window:.0f}s" + ) + _analyze_timestamps.append(now) + + +def _http_status_for_permission_error(exc: BaseException) -> HTTPStatus: + """Map PermissionError to auth (401), authorization (403), or capacity (429).""" + text = str(exc).lower().strip() + if any( + signal in text + for signal in ( + "rate limit", + "too many active jobs", + "quota exceeded", + "quota unavailable", + "budget", ) - _analyze_timestamps.append(now) + ): + return HTTPStatus.TOO_MANY_REQUESTS + if ( + text.startswith("oidc ") + or text.startswith("missing bearer") + or text.startswith("bearer token") + or text.startswith("service bearer") + or text.startswith("unsupported codex_audit_service_auth") + or "signature verification" in text + or "jwt segments" in text + or "signing key" in text + ): + return HTTPStatus.UNAUTHORIZED + return HTTPStatus.FORBIDDEN def _classify_service_failure(message: str) -> str: @@ -1600,7 +1642,7 @@ def _run_job(job_id: str, payload: dict[str, Any]) -> None: job["updated_at"] = _now() _write_job(job) _record_job_automation_run(job) - adapter = CursorAdapter() if payload.get("provider") == "cursor" else CodexAdapter() + adapter = resolve_execution_adapter(str(payload.get("provider") or "codex")) sandbox = _validate_sandbox(str(payload.get("sandbox") or "")) reasoning_effort = _resolve_codex_reasoning_effort(payload, str(payload.get("task") or TASK_EXECUTE)) execute_kwargs = { @@ -1646,7 +1688,10 @@ def _run_job(job_id: str, payload: dict[str, Any]) -> None: "provider": str(payload.get("provider") or "codex"), "model": str(payload.get("model") or ""), "reasoning_effort": str(reasoning_effort or ""), - "output": str(job.get("output") or ""), + "output_length": len(str(job.get("output") or "")), + "output_sha256": hashlib.sha256(str(job.get("output") or "").encode("utf-8")).hexdigest() + if job.get("output") + else "", "error": str(job.get("error") or ""), }, domain=str(job.get("domain") or ""), @@ -1778,19 +1823,24 @@ def do_GET(self) -> None: request_path = urlparse(self.path).path if request_path == "/healthz": + from service.cursor_account import subscription_research_readiness + health = get_health_monitor() + readiness = subscription_research_readiness() _json_response(self, HTTPStatus.OK, { "status": health.status, "uptime_seconds": health.uptime_seconds, "codex_research_routing": "v1", "subscription_research_routing": "v1", + "subscription_research_status": readiness.get("status"), + "subscription_research_reason": readiness.get("reason"), }) return if request_path == "/v1/ai/health": try: claims = authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return health = get_health_monitor() _json_response(self, HTTPStatus.OK, {"status": "ok", **health.snapshot()}) @@ -1799,7 +1849,7 @@ def do_GET(self) -> None: try: authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return org_health = read_org_health() _json_response(self, HTTPStatus.OK, org_health) @@ -1848,7 +1898,11 @@ def do_GET(self) -> None: except FileNotFoundError: _json_response(self, HTTPStatus.NOT_FOUND, {"status": "error", "error": "job not found"}) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response( + self, + _http_status_for_permission_error(exc), + {"status": "error", "error": str(exc)}, + ) except ValueError as exc: _json_response(self, HTTPStatus.BAD_REQUEST, {"status": "error", "error": str(exc)}) except Exception as exc: @@ -1909,7 +1963,11 @@ def do_POST(self) -> None: except PermissionError as exc: _audit_log("auth_error", path=self.path, error=str(exc)[:200]) - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response( + self, + _http_status_for_permission_error(exc), + {"status": "error", "error": str(exc)}, + ) except ValueError as exc: _json_response(self, HTTPStatus.BAD_REQUEST, {"status": "error", "error": str(exc)}) except Exception as exc: @@ -2129,7 +2187,7 @@ def _handle_execute_sync(self, claims: dict[str, Any], payload: dict[str, Any]) _json_response(self, HTTPStatus.TOO_MANY_REQUESTS, denial) return quota.record_execute(quota_repo, provider=str(payload.get("provider") or "codex")) - adapter = CursorAdapter() if payload.get("provider") == "cursor" else CodexAdapter() + adapter = resolve_execution_adapter(str(payload.get("provider") or "codex")) sandbox = _validate_sandbox(str(payload.get("sandbox") or "")) reasoning_effort = _resolve_codex_reasoning_effort(payload, str(payload.get("task") or TASK_EXECUTE)) execute_kwargs = { @@ -2257,6 +2315,7 @@ def _handle_review(self, claims: dict[str, Any], payload: dict[str, Any]) -> Non user=req.prompt, output=r.output, ).to_dict() + confidence = _extract_confidence_from_output(r.output) if r.success else None entry: dict[str, Any] = { "reviewer": r.provider, "model": r.model, @@ -2265,7 +2324,8 @@ def _handle_review(self, claims: dict[str, Any], payload: dict[str, Any]) -> Non "error": r.error if not r.success else "", "latency_seconds": r.latency_seconds, "usage": {"tokens_input": r.tokens_input, "tokens_output": r.tokens_output, "complete": r.usage_complete}, - "confidence": _extract_confidence_from_output(r.output) if r.success else 0.0, + "confidence": confidence if confidence is not None else 0.0, + "parse_ok": confidence is not None, "provenance_receipt": receipt, } results.append(entry) @@ -2280,19 +2340,24 @@ def _handle_review(self, claims: dict[str, Any], payload: dict[str, Any]) -> Non user=req.prompt, output=codex_result.output, ).to_dict() + codex_confidence = ( + _extract_confidence_from_output(codex_result.output) if codex_result.success else None + ) results.append({ "reviewer": "codex", "model": "codex-cli", "success": codex_result.success, "output": codex_result.output if codex_result.success else "", "error": codex_result.error if not codex_result.success else "", - "confidence": _extract_confidence_from_output(codex_result.output) if codex_result.success else 0.0, + "confidence": codex_confidence if codex_confidence is not None else 0.0, + "parse_ok": codex_confidence is not None, "provenance_receipt": receipt, }) # Step 4: compute consensus + recommended action consensus = _compute_consensus(results) all_ok = all(r["success"] for r in results) + parse_ok = all((not r["success"]) or bool(r.get("parse_ok")) for r in results) # Autonomy decision: confidence + file risk → recommended action repo = str(payload.get("source_repository") or "") @@ -2318,6 +2383,9 @@ def _handle_review(self, claims: dict[str, Any], payload: dict[str, Any]) -> Non result["provenance_receipt"]["policy_verdict"] == "eligible" for result in results ) + if not parse_ok: + consensus = "escalate" + policy_eligible = False if not policy_eligible: action = fail_closed_review_action(action) _audit_log("review_completed", consensus=consensus, all_success=all_ok, @@ -2341,7 +2409,7 @@ def _handle_automation_control(self) -> None: try: claims = authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return parsed = urlparse(self.path) params = parse_qs(parsed.query, keep_blank_values=True) @@ -2371,19 +2439,19 @@ def _handle_automation_control(self) -> None: manual_approval_id=manual_approval_id, ) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return if not _automation_operator_claims(claims): claims_repo = str(claims.get("repository") or "") if str(claims.get("auth_method") or "") == "static_token": if not _dashboard_repository_allowed(claims, repo): - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": "repo is not allowed"}) + _json_response(self, HTTPStatus.FORBIDDEN, {"status": "error", "error": "repo is not allowed"}) return elif repo and _dashboard_repository_allowed(claims, repo): pass elif repo and repo != claims_repo: if not manual_approval_valid: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": "repo is not allowed"}) + _json_response(self, HTTPStatus.FORBIDDEN, {"status": "error", "error": "repo is not allowed"}) return else: repo = claims_repo @@ -2442,7 +2510,7 @@ def _handle_list_automation_runs(self) -> None: try: claims = authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return parsed = urlparse(self.path) params = parse_qs(parsed.query) @@ -2464,7 +2532,7 @@ def _handle_get_automation_run(self, run_id: str) -> None: try: claims = authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return record = get_automation_run_ledger().get(run_id) if record is None: @@ -2473,7 +2541,7 @@ def _handle_get_automation_run(self, run_id: str) -> None: try: _assert_automation_run_access(record, claims) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return _json_response(self, HTTPStatus.OK, {"status": "ok", "run": record}) @@ -2690,14 +2758,14 @@ def _handle_get_change(self, change_id: str) -> None: except FileNotFoundError: _json_response(self, HTTPStatus.NOT_FOUND, {"status": "error", "error": "change not found"}) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) def _handle_list_changes(self) -> None: from urllib.parse import urlparse, parse_qs try: authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return parsed = urlparse(self.path) params = parse_qs(parsed.query) @@ -2714,7 +2782,7 @@ def _handle_effectiveness(self) -> None: try: authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return parsed = urlparse(self.path) params = parse_qs(parsed.query) @@ -2731,14 +2799,14 @@ def _handle_get_shadow(self) -> None: "disagreements": get_shadow_disagreements(), }) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) def _handle_quota_status(self) -> None: from urllib.parse import urlparse, parse_qs try: authenticate(self.headers, audience=DEFAULT_AUDIENCE) except PermissionError as exc: - _json_response(self, HTTPStatus.UNAUTHORIZED, {"status": "error", "error": str(exc)}) + _json_response(self, _http_status_for_permission_error(exc), {"status": "error", "error": str(exc)}) return parsed = urlparse(self.path) params = parse_qs(parsed.query) @@ -2763,44 +2831,83 @@ def _record_platform_execution_telemetry( logging.getLogger(__name__).warning("platform execution telemetry failed: %s", exc) -def _extract_confidence_from_output(output: str) -> float: +def _parse_review_json_object(output: str) -> dict[str, Any] | None: + """Parse exactly one JSON object from review output; reject multi-object/junk.""" + text = str(output or "").strip() + if not text: + return None + decoder = json.JSONDecoder() + try: + obj = json.loads(text) + if isinstance(obj, dict): + return obj + return None + except json.JSONDecodeError: + pass + start = text.find("{") + if start < 0: + return None + try: + obj, end = decoder.raw_decode(text, start) + except json.JSONDecodeError: + return None + if not isinstance(obj, dict): + return None + rest = text[end:].strip() + if "{" in rest: + return None + return obj + + +def _extract_confidence_from_output(output: str) -> float | None: """Extract confidence score from a reviewer's JSON output. - Looks for a ``confidence`` field (0.0–1.0) in the first JSON block found. - Returns 0.5 (neutral) if no confidence data is found. + Returns a finite float in [0.0, 1.0], or None when the payload is missing/invalid + so callers can fail closed instead of inventing a neutral 0.5. """ - if not output: - return 0.0 + obj = _parse_review_json_object(output) + if obj is None or "confidence" not in obj: + return None try: - match = re.search(r"\{[\s\S]*\}", output) - if match: - obj = json.loads(match.group(0)) - c = float(obj.get("confidence", 0.5)) - return max(0.0, min(1.0, c)) - except (json.JSONDecodeError, KeyError, TypeError, ValueError): - pass - return 0.5 + raw = obj["confidence"] + if isinstance(raw, bool): + return None + confidence = float(raw) + except (TypeError, ValueError): + return None + if not math.isfinite(confidence) or not 0.0 <= confidence <= 1.0: + return None + return confidence def _compute_consensus(results: list[dict[str, Any]]) -> str: """Simple consensus from review results — extracts approve/reject/escalate from JSON outputs.""" verdicts: list[str] = [] + parse_failures = 0 for r in results: if not r.get("success") or not r.get("output"): continue - try: - text = r["output"] - match = re.search(r"\{[\s\S]*\}", text) - if match: - obj = json.loads(match.group(0)) - verdict = str(obj.get("verdict", "")).lower() - if verdict in {"approve", "reject", "escalate", "verified", "mismatch", "agree", "review", "data_insufficient"}: - verdicts.append(verdict) - except (json.JSONDecodeError, KeyError): + obj = _parse_review_json_object(str(r["output"])) + if obj is None: + parse_failures += 1 continue + verdict = str(obj.get("verdict", "")).lower() + if verdict in { + "approve", + "reject", + "escalate", + "verified", + "mismatch", + "agree", + "review", + "data_insufficient", + }: + verdicts.append(verdict) + else: + parse_failures += 1 - if not verdicts: - return "unknown" + if parse_failures or not verdicts: + return "escalate" if all(v == verdicts[0] for v in verdicts): return verdicts[0] if any(v in {"reject", "mismatch"} for v in verdicts): diff --git a/service/autonomy.py b/service/autonomy.py index 0b29e7f3..bc35d286 100644 --- a/service/autonomy.py +++ b/service/autonomy.py @@ -16,20 +16,24 @@ medium — report generators, helper scripts, params low — docs, tests, README -Decision matrix:: +Decision matrix (matches DEFAULT_DECISION_MATRIX):: Confidence → -Risk ↓ <0.60 0.60-0.79 0.80-0.94 ≥0.95 +Risk ↓ <0.60 0.60–0.69 0.70–0.84 ≥0.85 ─────── ─────────── ──────────── ──────────── ─────────── - low auto_pr auto_merge auto_merge auto_merge - medium escalate auto_pr auto_pr auto_pr - high escalate escalate auto_pr auto_pr + low auto_pr auto_merge¹ auto_merge auto_merge + medium escalate escalate auto_pr auto_pr + high escalate escalate escalate auto_pr critical escalate escalate escalate escalate + +¹ low-risk auto_merge begins at confidence ≥ 0.60. +Empty ``changed_paths`` is treated as medium risk (never auto-merge). """ from __future__ import annotations import json +import math import os import re from dataclasses import dataclass, field @@ -255,9 +259,10 @@ def classify_changes_risk(changed_paths: list[str], *, policy: dict[str, Any] | """Classify the overall risk of a set of changed file paths. Returns the highest risk tier among all changed files. + Empty path lists are medium (unknown surface) so they cannot auto-merge. """ if not changed_paths: - return RISK_LOW + return RISK_MEDIUM tiers = {classify_file_risk(p, policy=policy) for p in changed_paths} for tier in (RISK_CRITICAL, RISK_HIGH, RISK_MEDIUM, RISK_LOW): if tier in tiers: @@ -351,17 +356,21 @@ def extract_confidence(verdicts: list[dict[str, Any]]) -> float: """Extract an aggregated confidence score from a list of reviewer verdicts. Each verdict dict may contain a ``confidence`` field (0.0–1.0). - Returns the weighted average, or 0.5 if no confidence data. + Returns the average of finite scores, or 0.0 when none are usable (fail-closed). """ scores: list[float] = [] for v in verdicts: + if "confidence" not in v: + continue try: - c = float(v.get("confidence", 0.5)) + c = float(v["confidence"]) + if not math.isfinite(c): + continue scores.append(max(0.0, min(1.0, c))) except (TypeError, ValueError): pass if not scores: - return 0.5 + return 0.0 return sum(scores) / len(scores) diff --git a/service/briefing_consumer.py b/service/briefing_consumer.py index 0b7d5a62..850652d0 100644 --- a/service/briefing_consumer.py +++ b/service/briefing_consumer.py @@ -417,9 +417,20 @@ def unavailable(reason: str) -> dict[str, Any]: if not source_repository: return unavailable("source_repository_required") try: + from service.provider_scenarios import ( + SCENARIO_DAILY_BRIEFING, + resolve_execute_kwargs, + ) + response = AiGatewayClient(config).execute( - prompt, task="daily_briefing", mode="review_only", research_stage="research_summary", - allowed_providers=list(config.research_providers), timeout=300, source_repository=source_repository, + prompt, + task="daily_briefing", + **resolve_execute_kwargs( + SCENARIO_DAILY_BRIEFING, + research_providers=config.research_providers, + ), + timeout=300, + source_repository=source_repository, ) except Exception: return unavailable("summary_execution_unavailable") diff --git a/service/contracts.py b/service/contracts.py index 77cb06b9..58ea5880 100644 --- a/service/contracts.py +++ b/service/contracts.py @@ -26,6 +26,15 @@ PROVIDER_OPENAI = "openai" PROVIDER_ANTHROPIC = "anthropic" PROVIDER_CODEX = "codex" +PROVIDER_CURSOR = "cursor" + +# CLI execution providers admitted by ExecuteRequest.allowed_providers. +EXECUTION_PROVIDERS = frozenset({PROVIDER_CODEX, PROVIDER_CURSOR}) +ALLOWED_EXECUTION_PROVIDER_CHAINS = ( + [PROVIDER_CODEX], + [PROVIDER_CURSOR], + [PROVIDER_CODEX, PROVIDER_CURSOR], +) @dataclass(frozen=True) @@ -62,9 +71,9 @@ class ExecuteRequest: def validate(self) -> None: if not self.prompt.strip(): raise ValueError("prompt must be a non-empty string") - if self.allowed_providers not in (["codex"], ["cursor"], ["codex", "cursor"]): + if self.allowed_providers not in ALLOWED_EXECUTION_PROVIDER_CHAINS: raise ValueError("allowed_providers must be [codex], [cursor], or [codex, cursor]") - if "cursor" in self.allowed_providers and self.mode != MODE_REVIEW_ONLY: + if PROVIDER_CURSOR in self.allowed_providers and self.mode != MODE_REVIEW_ONLY: raise ValueError("Cursor only supports review_only") if self.mode not in SUPPORTED_MODES: raise ValueError(f"mode must be one of {sorted(SUPPORTED_MODES)}") diff --git a/service/cursor_account.py b/service/cursor_account.py index 71898470..0a49a94b 100644 --- a/service/cursor_account.py +++ b/service/cursor_account.py @@ -1,6 +1,9 @@ -"""Cursor account roster and operator-reviewed subscription research admission. +"""Cursor subscription research admission (same economic class as Codex). -CLI discovery proves model availability, never remaining quota or paid usage. +Cursor CLI execution is subscription capacity, not OpenAI/Anthropic API billing. +Admission uses a trusted operator policy (models, daily call cap, on-demand off) +plus a fresh CLI roster. API analyze/review budgets must not gate or charge +these calls; Cursor dollar cost remains unknown and must not be reported as 0. """ from __future__ import annotations @@ -29,26 +32,67 @@ def _finite(value: Any) -> bool: def cursor_research_route(payload: dict[str, Any], usage: dict[str, Any], *, now: float) -> dict[str, Any]: deferred = {'action': 'defer', 'provider': 'cursor', 'reason': 'cursor_research_unavailable'} - if os.environ.get('AI_GATEWAY_CURSOR_ENABLED', '').lower() != 'true': - return {**deferred, 'reason': 'cursor_disabled'} try: + from service.automation_decision import _read_trusted_policy_file + policy_path = Path(os.environ['AI_GATEWAY_CURSOR_POLICY_PATH']) - if policy_path.is_symlink(): - return deferred - policy = json.loads(policy_path.read_text()) + raw, read_error = _read_trusted_policy_file(policy_path) + if read_error: + return {**deferred, 'reason': 'cursor_policy_untrusted'} + policy = json.loads(raw) roster = load_catalog(catalog_path()).subscription_rosters.get('cursor', {}) except (OSError, KeyError, TypeError, ValueError): return deferred return resolve_cursor_route(payload, usage, policy=policy, roster=roster, now=now) +def subscription_research_readiness(*, now: float | None = None) -> dict[str, str]: + """Desensitized Cursor subscription-lane readiness for health surfaces. + + Keeps the protocol capability field stable; readiness is a separate status. + Never proves remaining quota dollars. Admission follows request selection + plus trusted subscription policy and roster freshness—same shape as Codex. + """ + clock = time.time() if now is None else now + try: + from service.automation_decision import _read_trusted_policy_file + + policy_path = Path(os.environ['AI_GATEWAY_CURSOR_POLICY_PATH']) + raw, read_error = _read_trusted_policy_file(policy_path) + if read_error: + return {'status': 'not_ready', 'reason': 'cursor_policy_untrusted'} + policy = json.loads(raw) + roster = load_catalog(catalog_path()).subscription_rosters.get('cursor', {}) + except (OSError, KeyError, TypeError, ValueError): + return {'status': 'not_ready', 'reason': 'cursor_subscription_policy_unavailable'} + probe = resolve_cursor_route( + { + 'research_stage': 'research_summary', + 'complexity': 'low', + 'mode': 'review_only', + 'model': '', + 'reasoning_effort': 'auto', + }, + {'cursor_calls': 0}, + policy=policy, + roster=roster, + now=clock, + ) + if probe.get('action') == 'run': + return {'status': 'ready', 'reason': 'cursor_route_ready'} + return {'status': 'not_ready', 'reason': str(probe.get('reason') or 'cursor_research_unavailable')} + + + def resolve_cursor_route(payload, usage, *, policy, roster, now): deferred = {'action': 'defer', 'provider': 'cursor', 'reason': 'cursor_research_unavailable'} if not isinstance(policy, dict) or not isinstance(roster, dict) or not _finite(now): return deferred until = policy.get('valid_until') + # on_demand_disabled_verified: operator confirmed subscription will not + # overflow into Cursor paid on-demand. This is not an API-key budget gate. if (policy.get('on_demand_disabled_verified') is not True or not _finite(until) or until <= now): - return {**deferred, 'reason': 'cursor_spend_policy_unavailable'} + return {**deferred, 'reason': 'cursor_subscription_policy_unavailable'} updated = roster.get('updated_at') if (roster.get('status') != 'available' or roster.get('source') != 'cursor_cli_account' or not _finite(updated) or not 0 <= now - updated <= 86400): @@ -59,6 +103,9 @@ def resolve_cursor_route(payload, usage, *, policy, roster, now): return {**deferred, 'reason': 'cursor_capacity_unavailable'} stage = payload.get('research_stage') complexity = payload.get('complexity') or 'low' + # Cursor is subscription canary only for advisory stages; never promotion/codegen. + if stage not in {'research_summary', 'drift_analysis'}: + return {**deferred, 'reason': 'cursor_stage_not_canary'} if stage not in _STAGE_LEVELS or complexity not in ('low', 'medium', 'high') or payload.get('mode', 'review_only') != 'review_only': return deferred # Same deterministic stage/complexity floors as Codex, without assuming diff --git a/service/dual_review_primary.py b/service/dual_review_primary.py index 9f3cce8e..b103773f 100644 --- a/service/dual_review_primary.py +++ b/service/dual_review_primary.py @@ -12,6 +12,7 @@ from client.gateway_client import AiGatewayClient from service.dual_review import VERDICT_FAIL, VERDICT_INVALID, VERDICT_PASS, VERDICT_UNAVAILABLE, extract_verdict from service.dual_review_secondary import parse_llm_review_output +from service.provider_scenarios import SCENARIO_PROMOTION_PRIMARY_REVIEW, resolve_execute_kwargs _PRIMARY_SYSTEM = ( "You are the primary Codex reviewer for quantitative strategy promotion, risk, and recovery decisions. " @@ -164,9 +165,11 @@ def unavailable(reason: str, *, verdict: str = VERDICT_UNAVAILABLE) -> dict[str, if not 1 <= timeout <= 60: return unavailable("research_primary_invalid_timeout") result = AiGatewayClient(config).execute( - f"{_PRIMARY_SYSTEM}\n\n{prompt}", task="dual_review", mode="review_only", - research_stage="promotion_review", reasoning_effort="xhigh", allowed_providers=["codex"], - source_repository=source_repository, timeout=timeout * 60, + f"{_PRIMARY_SYSTEM}\n\n{prompt}", + task="dual_review", + **resolve_execute_kwargs(SCENARIO_PROMOTION_PRIMARY_REVIEW), + source_repository=source_repository, + timeout=timeout * 60, ) except Exception: return unavailable("research_primary_unavailable") diff --git a/service/provider_scenarios.py b/service/provider_scenarios.py new file mode 100644 index 00000000..f2f38bd4 --- /dev/null +++ b/service/provider_scenarios.py @@ -0,0 +1,393 @@ +"""Named AI call scenarios: adaptive defaults and forced overrides. + +Two selection modes for execute consumers: + +* Adaptive (default): scenario fills ``mode`` / ``research_stage`` / + ``allowed_providers`` (and optional complexity / pinned effort). Empty + model and effort stay omitted so the gateway subscription admission picks + them. ``AI_GATEWAY_RESEARCH_PROVIDERS`` only soft-selects Cursor on + ``cursor_eligible`` scenarios; fixed Codex lanes ignore that env. +* Forced: callers may pass ``allowed_providers``, ``model``, + ``reasoning_effort``, or ``complexity``. Incompatible values raise + ``ValueError``—never silently rewritten onto another provider or stage. + +Codex and Cursor are subscription CLI lanes. OpenAI/Anthropic analyze/review +are separate API-budget scenarios and are not substitutes for either CLI. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + +from service.contracts import ( + MODE_REVIEW_AND_FIX, + MODE_REVIEW_ONLY, + PROVIDER_CODEX, + PROVIDER_CURSOR, +) + +ENDPOINT_EXECUTE = "execute" +ENDPOINT_ANALYZE = "analyze" +ENDPOINT_REVIEW = "review" + +POLICY_FIXED_CODEX = "fixed_codex" +POLICY_RESEARCH_ENV = "research_env" +POLICY_API = "api" + +SCENARIO_RESEARCH_TASK_DIAGNOSIS = "research_task_diagnosis" +SCENARIO_PORTFOLIO_PROPOSAL_DIAGNOSIS = "portfolio_proposal_diagnosis" +SCENARIO_DAILY_BRIEFING = "daily_briefing" +SCENARIO_RESEARCH_SUMMARY = "research_summary" +SCENARIO_PROMOTION_PRIMARY_REVIEW = "promotion_primary_review" +SCENARIO_PLATFORM_BUGFIX = "platform_bugfix" +SCENARIO_SOXL_RSI2_CODEGEN = "soxl_rsi2_codegen" +SCENARIO_GLOBAL_ETF_CODEGEN = "global_etf_codegen" +SCENARIO_CN_INDEX_ETF_RESEARCH = "cn_index_etf_research" +SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE = "semantic_quality_acceptance" +SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS = "account_operational_diagnosis" +SCENARIO_SOXL_MANUAL_LEARNING = "soxl_manual_learning" +SCENARIO_NEW_RESEARCH_DESIGN = "new_research_design" +SCENARIO_API_ANALYZE = "api_analyze" +SCENARIO_API_DUAL_REVIEW = "api_dual_review" + +_VALID_COMPLEXITY = frozenset({"low", "medium", "high"}) +_VALID_EFFORTS = frozenset({"low", "medium", "high", "xhigh"}) + + +@dataclass(frozen=True) +class ProviderScenario: + scenario_id: str + endpoint: str + mode: str + research_stage: str + provider_policy: str + cursor_eligible: bool + fallback_chain_allowed: bool + purpose: str + default_complexity: str = "" + pin_reasoning_effort: str = "" + pin_model: str = "" + + +SCENARIOS: Mapping[str, ProviderScenario] = { + SCENARIO_RESEARCH_TASK_DIAGNOSIS: ProviderScenario( + scenario_id=SCENARIO_RESEARCH_TASK_DIAGNOSIS, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="drift_analysis", + provider_policy=POLICY_RESEARCH_ENV, + cursor_eligible=True, + fallback_chain_allowed=True, + purpose="Watcher-bound advisory diagnosis; no experiment or promotion authority.", + default_complexity="medium", + ), + SCENARIO_PORTFOLIO_PROPOSAL_DIAGNOSIS: ProviderScenario( + scenario_id=SCENARIO_PORTFOLIO_PROPOSAL_DIAGNOSIS, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="drift_analysis", + provider_policy=POLICY_RESEARCH_ENV, + cursor_eligible=True, + fallback_chain_allowed=True, + purpose="Portfolio research proposal advisory; same Cursor canary class as task diagnosis.", + default_complexity="medium", + ), + SCENARIO_DAILY_BRIEFING: ProviderScenario( + scenario_id=SCENARIO_DAILY_BRIEFING, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="research_summary", + provider_policy=POLICY_RESEARCH_ENV, + cursor_eligible=True, + fallback_chain_allowed=True, + purpose="Daily count summary advisory; expand only after diagnosis canary passes.", + default_complexity="low", + ), + SCENARIO_RESEARCH_SUMMARY: ProviderScenario( + scenario_id=SCENARIO_RESEARCH_SUMMARY, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="research_summary", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Standalone research summary scripts; stay Codex until separately authorized.", + default_complexity="low", + ), + SCENARIO_PROMOTION_PRIMARY_REVIEW: ProviderScenario( + scenario_id=SCENARIO_PROMOTION_PRIMARY_REVIEW, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="promotion_review", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Dual-review primary stays Codex-only even when research env lists Cursor.", + default_complexity="high", + pin_reasoning_effort="xhigh", + ), + SCENARIO_PLATFORM_BUGFIX: ProviderScenario( + scenario_id=SCENARIO_PLATFORM_BUGFIX, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Platform bugfix; Codex only; review_and_fix needs explicit approval path.", + default_complexity="medium", + ), + SCENARIO_SOXL_RSI2_CODEGEN: ProviderScenario( + scenario_id=SCENARIO_SOXL_RSI2_CODEGEN, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="optimization", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Bounded SOXL research codegen; Codex-only review_only optimization.", + default_complexity="high", + ), + SCENARIO_GLOBAL_ETF_CODEGEN: ProviderScenario( + scenario_id=SCENARIO_GLOBAL_ETF_CODEGEN, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="optimization", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Bounded global ETF research codegen; Codex-only.", + default_complexity="high", + ), + SCENARIO_CN_INDEX_ETF_RESEARCH: ProviderScenario( + scenario_id=SCENARIO_CN_INDEX_ETF_RESEARCH, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="optimization", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="CN index ETF research dispatch; Codex-only.", + default_complexity="high", + ), + SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE: ProviderScenario( + scenario_id=SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="drift_analysis", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Offline/service acceptance probe stays Codex; not a Cursor canary.", + default_complexity="medium", + ), + SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS: ProviderScenario( + scenario_id=SCENARIO_ACCOUNT_OPERATIONAL_DIAGNOSIS, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="drift_analysis", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Account operational diagnosis; Codex-only advisory.", + default_complexity="high", + ), + SCENARIO_SOXL_MANUAL_LEARNING: ProviderScenario( + scenario_id=SCENARIO_SOXL_MANUAL_LEARNING, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="optimization", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="SOXL manual learning advisory; Codex-only optimization.", + default_complexity="medium", + ), + SCENARIO_NEW_RESEARCH_DESIGN: ProviderScenario( + scenario_id=SCENARIO_NEW_RESEARCH_DESIGN, + endpoint=ENDPOINT_EXECUTE, + mode=MODE_REVIEW_ONLY, + research_stage="optimization", + provider_policy=POLICY_FIXED_CODEX, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="New-research design choice; Codex-only.", + default_complexity="low", + ), + SCENARIO_API_ANALYZE: ProviderScenario( + scenario_id=SCENARIO_API_ANALYZE, + endpoint=ENDPOINT_ANALYZE, + mode="", + research_stage="", + provider_policy=POLICY_API, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Sync OpenAI/Anthropic analyze; separate API budget; never Cursor substitute.", + ), + SCENARIO_API_DUAL_REVIEW: ProviderScenario( + scenario_id=SCENARIO_API_DUAL_REVIEW, + endpoint=ENDPOINT_REVIEW, + mode=MODE_REVIEW_ONLY, + research_stage="", + provider_policy=POLICY_API, + cursor_eligible=False, + fallback_chain_allowed=False, + purpose="Multi-model API review with optional Codex verifier; not Cursor.", + ), +} + + +def get_scenario(scenario_id: str) -> ProviderScenario: + scenario = SCENARIOS.get(str(scenario_id or "").strip()) + if scenario is None: + raise ValueError(f"unknown provider scenario: {scenario_id!r}") + return scenario + + +def _normalize_research_providers(research_providers: tuple[str, ...] | list[str]) -> tuple[str, ...]: + providers = tuple(str(item).strip() for item in research_providers if str(item).strip()) + if providers not in ((PROVIDER_CODEX,), (PROVIDER_CURSOR,), (PROVIDER_CODEX, PROVIDER_CURSOR)): + raise ValueError("research_providers must be (codex,), (cursor,), or (codex, cursor)") + return providers + + +def _normalize_forced_providers(allowed_providers: list[str] | tuple[str, ...]) -> list[str]: + providers = [str(item).strip() for item in allowed_providers if str(item).strip()] + if providers not in ([PROVIDER_CODEX], [PROVIDER_CURSOR], [PROVIDER_CODEX, PROVIDER_CURSOR]): + raise ValueError("allowed_providers must be [codex], [cursor], or [codex, cursor]") + return providers + + +def resolve_allowed_providers( + scenario_id: str, + *, + research_providers: tuple[str, ...] | list[str] = (PROVIDER_CODEX,), + allowed_providers: list[str] | tuple[str, ...] | None = None, +) -> list[str]: + """Resolve execution provider chain for a named scenario. + + Adaptive: soft research_providers env on cursor_eligible scenarios. + Forced: explicit allowed_providers must be legal for the scenario or raise. + """ + scenario = get_scenario(scenario_id) + if scenario.provider_policy == POLICY_API: + raise ValueError(f"{scenario_id} uses API endpoint, not allowed_providers") + + if allowed_providers is not None: + forced = _normalize_forced_providers(allowed_providers) + if PROVIDER_CURSOR in forced: + if not scenario.cursor_eligible: + raise ValueError(f"{scenario_id} does not allow Cursor") + if forced == [PROVIDER_CODEX, PROVIDER_CURSOR] and not scenario.fallback_chain_allowed: + raise ValueError(f"{scenario_id} does not allow Codex→Cursor fallback chain") + if scenario.provider_policy == POLICY_FIXED_CODEX and forced != [PROVIDER_CODEX]: + raise ValueError(f"{scenario_id} requires allowed_providers=['codex']") + return forced + + if scenario.provider_policy == POLICY_FIXED_CODEX or not scenario.cursor_eligible: + return [PROVIDER_CODEX] + providers = _normalize_research_providers(research_providers) + if providers == (PROVIDER_CODEX, PROVIDER_CURSOR) and not scenario.fallback_chain_allowed: + raise ValueError(f"{scenario_id} does not allow Codex→Cursor fallback chain") + return list(providers) + + +def resolve_execute_kwargs( + scenario_id: str, + *, + research_providers: tuple[str, ...] | list[str] = (PROVIDER_CODEX,), + mode: str | None = None, + allowed_providers: list[str] | tuple[str, ...] | None = None, + model: str | None = None, + reasoning_effort: str | None = None, + complexity: str | None = None, + research_stage: str | None = None, +) -> dict[str, Any]: + """Build execute() kwargs: adaptive scenario defaults plus optional forced pins.""" + scenario = get_scenario(scenario_id) + if scenario.endpoint != ENDPOINT_EXECUTE: + raise ValueError(f"{scenario_id} is not an execute scenario") + + effective_mode = MODE_REVIEW_ONLY if mode is None else str(mode).strip().lower() + if scenario.scenario_id == SCENARIO_PLATFORM_BUGFIX: + if effective_mode not in {MODE_REVIEW_ONLY, MODE_REVIEW_AND_FIX}: + raise ValueError("platform_bugfix requires review_only or review_and_fix") + elif effective_mode != scenario.mode: + raise ValueError(f"{scenario_id} requires mode={scenario.mode}") + + allowed = resolve_allowed_providers( + scenario_id, + research_providers=research_providers, + allowed_providers=allowed_providers, + ) + if PROVIDER_CURSOR in allowed and effective_mode != MODE_REVIEW_ONLY: + raise ValueError("Cursor scenarios require review_only") + + kwargs: dict[str, Any] = { + "mode": effective_mode, + "allowed_providers": allowed, + } + effective_stage = scenario.research_stage + if research_stage not in (None, ""): + if scenario.provider_policy != POLICY_FIXED_CODEX: + raise ValueError("research_stage override requires a fixed_codex scenario") + override = str(research_stage).strip() + if override not in {"research_summary", "drift_analysis", "optimization", "promotion_review"}: + raise ValueError("unsupported research_stage override") + effective_stage = override + if effective_stage: + kwargs["research_stage"] = effective_stage + if PROVIDER_CURSOR in allowed and effective_stage not in cursor_canary_research_stages(): + raise ValueError("Cursor is only allowed for canary research stages") + + effective_complexity = ( + str(complexity).strip().lower() + if complexity not in (None, "") + else scenario.default_complexity + ) + if effective_complexity: + if effective_complexity not in _VALID_COMPLEXITY: + raise ValueError("complexity must be low, medium, or high") + kwargs["complexity"] = effective_complexity + + if reasoning_effort not in (None, "", "auto"): + effort = str(reasoning_effort).strip().lower() + if effort not in _VALID_EFFORTS: + raise ValueError("reasoning_effort must be low, medium, high, or xhigh") + if scenario.pin_reasoning_effort and effort != scenario.pin_reasoning_effort: + raise ValueError( + f"{scenario_id} requires reasoning_effort={scenario.pin_reasoning_effort}" + ) + kwargs["reasoning_effort"] = effort + elif scenario.pin_reasoning_effort: + kwargs["reasoning_effort"] = scenario.pin_reasoning_effort + + if model not in (None, "", "auto"): + forced_model = str(model).strip() + if not forced_model: + raise ValueError("model must be non-empty when forced") + if scenario.pin_model and forced_model != scenario.pin_model: + raise ValueError(f"{scenario_id} requires model={scenario.pin_model}") + kwargs["model"] = forced_model + elif scenario.pin_model: + kwargs["model"] = scenario.pin_model + + return kwargs + + +def cursor_canary_scenarios() -> tuple[str, ...]: + return tuple( + scenario.scenario_id + for scenario in SCENARIOS.values() + if scenario.cursor_eligible and scenario.endpoint == ENDPOINT_EXECUTE + ) + + +def cursor_canary_research_stages() -> frozenset[str]: + """Research stages that may select Cursor (gateway + consumer enforce).""" + return frozenset( + scenario.research_stage + for scenario in SCENARIOS.values() + if scenario.cursor_eligible and scenario.research_stage + ) diff --git a/service/quota.py b/service/quota.py index eef7ee9e..4fa671ae 100644 --- a/service/quota.py +++ b/service/quota.py @@ -553,7 +553,11 @@ def cursor_usage(self) -> dict[str, Any]: )} def record_execute(self, repo: str, *, provider: str = "codex") -> None: - """Record subscription calls separately; Cursor cost is unavailable.""" + """Record subscription CLI calls; never charge OpenAI/Anthropic API budget. + + Codex keeps a nominal accounting estimate. Cursor increments call count + only—subscription dollars are unknown and must stay cost_incomplete. + """ if provider not in {"codex", "cursor"}: raise ValueError("unsupported execution provider") cost = DEFAULT_MODEL_COSTS.get("codex-cli", {}).get("flat", 0.05) diff --git a/tests/test_ai_gateway_service_execute_telemetry.py b/tests/test_ai_gateway_service_execute_telemetry.py index f0c3f57d..0ac72745 100644 --- a/tests/test_ai_gateway_service_execute_telemetry.py +++ b/tests/test_ai_gateway_service_execute_telemetry.py @@ -135,20 +135,21 @@ def test_research_routes_select_supported_model_and_defer_before_provider(self) patch.object(gateway, "get_health_monitor"), patch.object(gateway, "_json_response") as response, patch.object(gateway, "_submit_job", return_value={"job_id": "synthetic"}) as submit, - patch.object(gateway, "CodexAdapter") as adapter, + patch.object(gateway, "resolve_execution_adapter") as resolve, ): - adapter.return_value.execute.return_value = SimpleNamespace(success=True, output="synthetic", error="") + adapter = resolve.return_value + adapter.execute.return_value = SimpleNamespace(success=True, output="synthetic", error="") getattr(gateway.AiGatewayRequestHandler, method)(object(), {"repository": "Synthetic/caller"}, payload) if used == 65: self.assertEqual(response.call_args.args[1], 429) self.assertEqual(response.call_args.args[2]["status"], "deferred") self.assertEqual(response.call_args.args[2]["retry_at"], 9000) submit.assert_not_called() - adapter.assert_not_called() + resolve.assert_not_called() record.assert_not_called() else: self.assertIn(response.call_args.args[1], (200, 202)) - selected = submit.call_args.args[1] if method.endswith("async") else adapter.return_value.execute.call_args.kwargs + selected = submit.call_args.args[1] if method.endswith("async") else adapter.execute.call_args.kwargs self.assertEqual(selected["model"], "gpt-5.6-sol") self.assertEqual(selected["reasoning_effort"], "high") record.assert_called_once() @@ -163,14 +164,14 @@ def test_exhausted_subscription_rejects_both_execute_routes_before_submission(se patch.object(gateway, "get_quota_manager", return_value=quota), patch.object(gateway, "_json_response") as response, patch.object(gateway, "_submit_job") as submit, - patch.object(gateway, "CodexAdapter") as adapter, + patch.object(gateway, "resolve_execution_adapter") as resolve, ): getattr(gateway.AiGatewayRequestHandler, method)( object(), {"repository": "Synthetic/caller"}, {"prompt": "synthetic", "mode": "review_only"}, ) self.assertEqual(response.call_args.args[1], 429) submit.assert_not_called() - adapter.assert_not_called() + resolve.assert_not_called() self.assertNotIn("Synthetic/caller", quota._records) def test_failed_codex_diagnostics_never_reach_job_or_telemetry(self) -> None: @@ -205,10 +206,10 @@ def test_failed_codex_diagnostics_never_reach_job_or_telemetry(self) -> None: @patch("service.ai_gateway_service.get_health_monitor") @patch("service.ai_gateway_service._record_job_automation_run") @patch("service.ai_gateway_service._audit_log") - @patch("service.ai_gateway_service.CodexAdapter.execute") + @patch("service.ai_gateway_service.resolve_execution_adapter") def test_run_job_records_execution_telemetry( self, - mock_execute, + mock_resolve, _mock_audit_log, _mock_record_job_automation_run, mock_health_monitor, @@ -230,7 +231,7 @@ def _read_job(_job_id: str) -> dict[str, object]: def _write_job(payload: dict[str, object]) -> None: writes.append(dict(payload)) - mock_execute.return_value = SimpleNamespace(success=True, output="done", error="") + mock_resolve.return_value.execute.return_value = SimpleNamespace(success=True, output="done", error="") health = mock_health_monitor.return_value health.record.return_value = None @@ -250,14 +251,18 @@ def _write_job(payload: dict[str, object]) -> None: self.assertEqual(execution_result["status"], "succeeded") self.assertEqual(execution_result["model"], "gpt-5.4-mini") self.assertEqual(call_domain, "cn_equity") + self.assertNotIn("output", execution_result) + self.assertEqual(execution_result["output_length"], 4) + self.assertEqual(len(execution_result["output_sha256"]), 64) def test_run_job_disables_shell_tool_only_for_readonly_platform_bugfix(self) -> None: with patch.object(gateway, "_record_job_automation_run"), patch.object(gateway, "_audit_log"), patch.object( gateway, "get_health_monitor" ), patch.object(gateway, "_record_platform_execution_telemetry"), patch.object( gateway, "_classify_failure", return_value="" - ), patch.object(gateway, "_write_job"), patch.object(gateway, "CodexAdapter") as adapter: - adapter.return_value.execute.return_value = SimpleNamespace(success=True, output="review", error="") + ), patch.object(gateway, "_write_job"), patch.object(gateway, "resolve_execution_adapter") as resolve: + adapter = resolve.return_value + adapter.execute.return_value = SimpleNamespace(success=True, output="review", error="") for task, mode, expected, expected_tools in ( ("platform_bugfix", "review_only", False, False), (" platform_bugfix ", " REVIEW_ONLY ", False, False), @@ -275,7 +280,7 @@ def test_run_job_disables_shell_tool_only_for_readonly_platform_bugfix(self) -> if task == "platform_bugfix" and mode == "review_and_fix" and not expected: payload["manual_approval_id"] = "sg-history-482" gateway._run_job("job-1", payload) - kwargs = adapter.return_value.execute.call_args.kwargs + kwargs = adapter.execute.call_args.kwargs self.assertEqual(kwargs["shell_tool_enabled"], expected) self.assertEqual(kwargs["tools_disabled"], expected_tools) @@ -284,8 +289,9 @@ def test_run_job_disables_all_codex_tools_for_readonly_drift_analysis(self) -> N gateway, "get_health_monitor" ), patch.object(gateway, "_record_platform_execution_telemetry"), patch.object( gateway, "_classify_failure", return_value="" - ), patch.object(gateway, "_write_job"), patch.object(gateway, "CodexAdapter") as adapter: - adapter.return_value.execute.return_value = SimpleNamespace(success=True, output="review", error="") + ), patch.object(gateway, "_write_job"), patch.object(gateway, "resolve_execution_adapter") as resolve: + adapter = resolve.return_value + adapter.execute.return_value = SimpleNamespace(success=True, output="review", error="") cases = [ ({"task": "execute", "mode": "review_only", "research_stage": "drift_analysis", "sandbox": "read-only"}, True), ({"mode": "review_only", "research_stage": "drift_analysis", "sandbox": "read-only"}, True), @@ -300,7 +306,7 @@ def test_run_job_disables_all_codex_tools_for_readonly_drift_analysis(self) -> N return_value={"job_id": "job-1", "status": "queued", **payload}, ): gateway._run_job("job-1", {"prompt": "review", **payload}) - kwargs = adapter.return_value.execute.call_args.kwargs + kwargs = adapter.execute.call_args.kwargs self.assertEqual(kwargs["shell_tool_enabled"], not expected_disabled) self.assertEqual(kwargs["tools_disabled"], expected_disabled) @@ -316,11 +322,12 @@ def test_sync_execute_disables_all_codex_tools_for_readonly_drift_analysis(self) patch.object(gateway, "_admit_codex_execute", return_value=None), patch.object(gateway, "get_health_monitor"), patch.object(gateway, "_json_response"), - patch.object(gateway, "CodexAdapter") as adapter, + patch.object(gateway, "resolve_execution_adapter") as resolve, ): - adapter.return_value.execute.return_value = SimpleNamespace(success=True, output="review", error="") + adapter = resolve.return_value + adapter.execute.return_value = SimpleNamespace(success=True, output="review", error="") gateway.AiGatewayRequestHandler._handle_execute_sync(object(), {"repository": "Synthetic/caller"}, payload) - kwargs = adapter.return_value.execute.call_args.kwargs + kwargs = adapter.execute.call_args.kwargs self.assertFalse(kwargs["shell_tool_enabled"]) self.assertTrue(kwargs["tools_disabled"]) diff --git a/tests/test_ai_gateway_service_get_routes.py b/tests/test_ai_gateway_service_get_routes.py index d3903a50..b9f0ef50 100644 --- a/tests/test_ai_gateway_service_get_routes.py +++ b/tests/test_ai_gateway_service_get_routes.py @@ -225,7 +225,7 @@ def test_automation_run_update_rejects_owner_mismatch_even_for_operator(self) -> ) with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(second_request, timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) finally: server.shutdown() server.server_close() @@ -294,7 +294,7 @@ def test_external_automation_run_update_cannot_overwrite_service_job_run(self) - ) with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(request, timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) self.assertEqual(get_automation_run_ledger().get("service-run")["task_state"], "running") finally: server.shutdown() @@ -404,7 +404,7 @@ def test_static_token_dashboard_can_query_allowlisted_control_repo(self) -> None ) with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(missing_repo_request, timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) finally: server.shutdown() server.server_close() @@ -500,7 +500,7 @@ def test_automation_control_rejects_cross_repository_query_for_oidc_callers(self f"{base_url}/v1/ai/automation/control?repo=QuantStrategyLab/OtherRepo", timeout=5, ) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) finally: server.shutdown() server.server_close() @@ -557,7 +557,7 @@ def test_automation_control_manual_grant_allows_only_exact_cross_repository_scop with patch("service.ai_gateway_service.authenticate", return_value=changed_claims): with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(f"{base_url}/v1/ai/automation/control?{query}", timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) with patch( "service.ai_gateway_service.authenticate", @@ -565,7 +565,7 @@ def test_automation_control_manual_grant_allows_only_exact_cross_repository_scop ): with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(f"{base_url}/v1/ai/automation/control?{query}", timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) with patch( "service.ai_gateway_service.load_execution_policy", @@ -573,13 +573,13 @@ def test_automation_control_manual_grant_allows_only_exact_cross_repository_scop ), patch("service.ai_gateway_service.authenticate", return_value=claims): with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(f"{base_url}/v1/ai/automation/control?{query}", timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) no_grant = urllib.parse.urlencode({"repo": approval["source_repository"], "mode": "review_and_fix"}) with patch("service.ai_gateway_service.authenticate", return_value=claims): with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(f"{base_url}/v1/ai/automation/control?{no_grant}", timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) finally: server.shutdown() server.server_close() @@ -1437,7 +1437,7 @@ def test_review_route_uses_service_owned_trusted_live_equivalent_proof(self) -> ) with self.assertRaises(urllib.error.HTTPError) as ctx: urllib.request.urlopen(missing_source_request, timeout=5) - self.assertEqual(ctx.exception.code, 401) + self.assertEqual(ctx.exception.code, 403) finally: server.shutdown() server.server_close() diff --git a/tests/test_autonomy.py b/tests/test_autonomy.py index 513b2110..dcd5a76d 100644 --- a/tests/test_autonomy.py +++ b/tests/test_autonomy.py @@ -55,10 +55,14 @@ def test_market_strategy_paths_are_high_risk(self) -> None: with self.subTest(path=path): self.assertEqual(classify_file_risk(path), RISK_HIGH) - def test_low_risk_high_confidence_can_auto_merge(self) -> None: - result = recommended_action([{"confidence": 0.96}], ["docs/README.md"]) - self.assertEqual(result["action"], ACTION_AUTO_MERGE) - self.assertEqual(result["risk"], RISK_LOW) + def test_empty_changed_paths_are_medium_and_cannot_auto_merge(self) -> None: + from service.autonomy import classify_changes_risk + + self.assertEqual(classify_changes_risk([]), RISK_MEDIUM) + result = recommended_action([{"confidence": 0.99}], []) + self.assertEqual(result["risk"], RISK_MEDIUM) + self.assertNotEqual(result["action"], ACTION_AUTO_MERGE) + self.assertFalse(result["auto_merge_allowed"]) def test_medium_risk_defaults_to_auto_pr_not_merge(self) -> None: self.assertEqual(recommended_action([{"confidence": 0.90}], ["scripts/build.py"])["action"], ACTION_AUTO_PR) diff --git a/tests/test_contracts.py b/tests/test_contracts.py index c3c07d6c..1c705292 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -34,6 +34,17 @@ def test_mode_constants(self) -> None: self.assertEqual(MODE_REVIEW_ONLY, "review_only") self.assertEqual(MODE_REVIEW_AND_FIX, "review_and_fix") + def test_provider_constants_include_cursor(self) -> None: + from service.contracts import ( + ALLOWED_EXECUTION_PROVIDER_CHAINS, + EXECUTION_PROVIDERS, + PROVIDER_CODEX, + PROVIDER_CURSOR, + ) + + self.assertEqual(EXECUTION_PROVIDERS, frozenset({PROVIDER_CODEX, PROVIDER_CURSOR})) + self.assertIn([PROVIDER_CODEX, PROVIDER_CURSOR], ALLOWED_EXECUTION_PROVIDER_CHAINS) + class TestAnalyzeRequest(unittest.TestCase): """AnalyzeRequest schema validation.""" diff --git a/tests/test_cursor_adapter.py b/tests/test_cursor_adapter.py index 4deac20a..04857e44 100644 --- a/tests/test_cursor_adapter.py +++ b/tests/test_cursor_adapter.py @@ -28,19 +28,24 @@ def run(command, **kwargs): assert command[command.index('--sandbox') + 1] == 'enabled' assert command[command.index('--model') + 1] == 'synthetic-model' assert command[command.index('--allowed-tools') + 1] == '' - assert '--exclude-workspace-context' in command + assert '--exclude-workspace-context' not in command assert '--disable-auto-update' in command assert not {'--force', '--yolo', '--approve-mcps', '--api-key'} & set(command) assert kwargs['input'].endswith('Task and supplied evidence:\nsynthetic prompt') assert 'OPENAI_API_KEY' not in kwargs['env'] assert 'CURSOR_API_KEY' not in kwargs['env'] + assert kwargs['env'].get('HOME') == '/synthetic/cursor-home' workspace = Path(kwargs['cwd']) assert workspace.joinpath('AGENTS.md').is_file() assert workspace.joinpath('.agents/skills/research-evidence/SKILL.md').is_file() assert 'Shell(*)' in json.loads(workspace.joinpath('.cursor/cli.json').read_text())['permissions']['deny'] return SimpleNamespace(returncode=0, stdout=json.dumps({'type': 'result', 'subtype': 'success', 'is_error': False, 'result': 'advisory'}), stderr='') with patch('service.adapters.cursor_adapter.shutil.which', return_value='/synthetic/agent'), patch.dict( - 'os.environ', {'OPENAI_API_KEY': 'synthetic-private', 'CURSOR_API_KEY': 'synthetic-private'} + 'os.environ', { + 'OPENAI_API_KEY': 'synthetic-private', + 'CURSOR_API_KEY': 'synthetic-private', + 'AI_GATEWAY_CURSOR_HOME': '/synthetic/cursor-home', + } ), patch('service.adapters.cursor_adapter.subprocess.run', side_effect=run) as runner: result = CursorAdapter().execute(prompt='synthetic prompt', model='synthetic-model', reasoning_effort='high') assert result.success and result.output == 'advisory' diff --git a/tests/test_dual_review_research_routing.py b/tests/test_dual_review_research_routing.py index d3ba32a5..38abcb03 100644 --- a/tests/test_dual_review_research_routing.py +++ b/tests/test_dual_review_research_routing.py @@ -45,6 +45,8 @@ def test_actual_pipeline_routes_research_primary_and_keeps_actual_codex_model(tr result = run_pipeline(trigger=trigger, strategy_profile="synthetic", context={}) assert execute.call_args.kwargs.get("research_stage") == "promotion_review" assert execute.call_args.kwargs["allowed_providers"] == ["codex"] + assert execute.call_args.kwargs["reasoning_effort"] == "xhigh" + assert execute.call_args.kwargs["complexity"] == "high" primary = result["primary_review"] assert primary["provider"] == "codex" and primary["model"] == "gpt-6-astra" assert primary["reasoning_effort"] == "xhigh" diff --git a/tests/test_provider_opt_contract.py b/tests/test_provider_opt_contract.py new file mode 100644 index 00000000..2335582c --- /dev/null +++ b/tests/test_provider_opt_contract.py @@ -0,0 +1,186 @@ +"""Provider-lane contract, adapter selection, Cursor policy trust, and limiter.""" + +from __future__ import annotations + +import json +import os +import stat +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest.mock import patch + +from client.config import ProviderConfig +from service.adapters.codex_adapter import CodexAdapter +from service.adapters.cursor_adapter import CursorAdapter +from service.adapters.execution import resolve_execution_adapter +from service.automation_decision import EXECUTION_POLICY_OWNER_ENV +from service.contracts import ( + ALLOWED_EXECUTION_PROVIDER_CHAINS, + EXECUTION_PROVIDERS, + ExecuteRequest, + MODE_REVIEW_AND_FIX, + MODE_REVIEW_ONLY, + PROVIDER_CODEX, + PROVIDER_CURSOR, +) +from service import ai_gateway_service as gateway +from service import cursor_account + + +class ProviderContractTests(unittest.TestCase): + def test_execution_provider_constants_include_cursor(self) -> None: + self.assertEqual(EXECUTION_PROVIDERS, frozenset({PROVIDER_CODEX, PROVIDER_CURSOR})) + self.assertIn([PROVIDER_CODEX], ALLOWED_EXECUTION_PROVIDER_CHAINS) + self.assertIn([PROVIDER_CURSOR], ALLOWED_EXECUTION_PROVIDER_CHAINS) + self.assertIn([PROVIDER_CODEX, PROVIDER_CURSOR], ALLOWED_EXECUTION_PROVIDER_CHAINS) + + def test_execute_request_accepts_documented_provider_chains(self) -> None: + for chain in ALLOWED_EXECUTION_PROVIDER_CHAINS: + mode = MODE_REVIEW_ONLY if PROVIDER_CURSOR in chain else MODE_REVIEW_AND_FIX + ExecuteRequest(prompt="synthetic", mode=mode, allowed_providers=list(chain)).validate() + + def test_execute_request_rejects_unknown_provider_chain(self) -> None: + with self.assertRaisesRegex(ValueError, "allowed_providers"): + ExecuteRequest(prompt="synthetic", allowed_providers=["openai"]).validate() + + def test_provider_config_exposes_cursor_factory(self) -> None: + config = ProviderConfig.cursor("composer-2.5") + self.assertEqual(config.label, "cursor") + self.assertEqual(config.model, "composer-2.5") + self.assertTrue(config.can_execute_code) + + +class ExecutionAdapterTests(unittest.TestCase): + def test_resolve_defaults_and_known_providers(self) -> None: + self.assertIsInstance(resolve_execution_adapter(""), CodexAdapter) + self.assertIsInstance(resolve_execution_adapter("codex"), CodexAdapter) + self.assertIsInstance(resolve_execution_adapter("cursor"), CursorAdapter) + + def test_resolve_unknown_provider_fail_closed(self) -> None: + with self.assertRaisesRegex(ValueError, "unsupported execution provider"): + resolve_execution_adapter("openai") + + +class CursorPolicyTrustTests(unittest.TestCase): + def _owner_env(self) -> dict[str, str]: + return {EXECUTION_POLICY_OWNER_ENV: f"{os.getuid()}:{os.getgid()}"} + + def test_symlink_policy_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as raw_dir: + root = Path(raw_dir) + target = root / "real.json" + target.write_text("{}", encoding="utf-8") + os.chmod(target, stat.S_IRUSR | stat.S_IWUSR) + link = root / "policy.json" + link.symlink_to(target) + with patch.dict( + os.environ, + { + **self._owner_env(), + "AI_GATEWAY_CURSOR_POLICY_PATH": str(link), + }, + clear=False, + ): + route = cursor_account.cursor_research_route( + {"research_stage": "research_summary", "mode": "review_only"}, + {"cursor_calls": 0}, + now=time.time(), + ) + self.assertEqual(route["action"], "defer") + self.assertEqual(route["reason"], "cursor_policy_untrusted") + + def test_readiness_reports_not_ready_when_policy_path_missing(self) -> None: + with patch.dict( + os.environ, + {"AI_GATEWAY_CURSOR_POLICY_PATH": "/tmp/missing-cursor-policy.json"}, + clear=False, + ): + readiness = cursor_account.subscription_research_readiness(now=time.time()) + self.assertEqual(readiness["status"], "not_ready") + self.assertIn(readiness["reason"], {"cursor_subscription_policy_unavailable", "cursor_policy_untrusted"}) + + def test_group_writable_policy_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as raw_dir: + policy_path = Path(raw_dir) / "policy.json" + policy_path.write_text("{}", encoding="utf-8") + policy_path.chmod(policy_path.stat().st_mode | stat.S_IWGRP) + with patch.dict( + os.environ, + { + **self._owner_env(), + "AI_GATEWAY_CURSOR_POLICY_PATH": str(policy_path), + }, + clear=False, + ): + route = cursor_account.cursor_research_route( + {"research_stage": "research_summary", "mode": "review_only"}, + {"cursor_calls": 0}, + now=time.time(), + ) + self.assertEqual(route["action"], "defer") + self.assertEqual(route["reason"], "cursor_policy_untrusted") + + def test_readiness_ready_when_trusted_route_admits(self) -> None: + now = time.time() + policy = { + "on_demand_disabled_verified": True, + "valid_until": now + 3600, + "max_daily_calls": 10, + "models": { + "composer-2": { + "quality_level": 0, + "supported_reasoning_efforts": ["low"], + } + }, + } + roster = { + "status": "available", + "source": "cursor_cli_account", + "updated_at": now, + "models": ["composer-2"], + } + catalog = type("Catalog", (), {"subscription_rosters": {"cursor": roster}})() + with patch.dict( + os.environ, + {**self._owner_env(), "AI_GATEWAY_CURSOR_POLICY_PATH": "/tmp/unused"}, + clear=False, + ), patch( + "service.automation_decision._read_trusted_policy_file", + return_value=(json.dumps(policy), ""), + ), patch("service.cursor_account.load_catalog", return_value=catalog): + readiness = cursor_account.subscription_research_readiness(now=now) + self.assertEqual(readiness, {"status": "ready", "reason": "cursor_route_ready"}) + + +class RateLimitLockTests(unittest.TestCase): + def test_rate_limit_cap_holds_under_concurrent_callers(self) -> None: + gateway._analyze_timestamps.clear() + accepted = 0 + rejected = 0 + lock = threading.Lock() + + def worker() -> None: + nonlocal accepted, rejected + try: + gateway._check_rate_limit(max_per_window=5, window=60.0) + with lock: + accepted += 1 + except PermissionError: + with lock: + rejected += 1 + + threads = [threading.Thread(target=worker) for _ in range(20)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + self.assertEqual(accepted, 5) + self.assertEqual(rejected, 15) + gateway._analyze_timestamps.clear() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_provider_opt_hardening.py b/tests/test_provider_opt_hardening.py new file mode 100644 index 00000000..02e68dde --- /dev/null +++ b/tests/test_provider_opt_hardening.py @@ -0,0 +1,80 @@ +"""Offline hardening coverage for remaining provider-opt audit items.""" + +from __future__ import annotations + +import unittest +from http import HTTPStatus +from pathlib import Path + +import service.ai_gateway_service as gateway + + +class TestStrictReviewJsonParse(unittest.TestCase): + def test_extract_confidence_requires_single_valid_object(self) -> None: + self.assertEqual( + gateway._extract_confidence_from_output('{"verdict":"approve","confidence":0.9}'), + 0.9, + ) + self.assertIsNone(gateway._extract_confidence_from_output("")) + self.assertIsNone(gateway._extract_confidence_from_output("not json")) + self.assertIsNone( + gateway._extract_confidence_from_output('{"verdict":"approve","confidence":0.9}{"confidence":0.1}') + ) + self.assertIsNone(gateway._extract_confidence_from_output('{"verdict":"approve"}')) + self.assertIsNone(gateway._extract_confidence_from_output('{"confidence":"NaN"}')) + self.assertIsNone(gateway._extract_confidence_from_output('{"confidence":1.5}')) + + def test_compute_consensus_fail_closed_on_junk(self) -> None: + self.assertEqual( + gateway._compute_consensus([{"success": True, "output": '{"verdict":"approve","confidence":0.9}'}]), + "approve", + ) + self.assertEqual( + gateway._compute_consensus([{"success": True, "output": "approve maybe"}]), + "escalate", + ) + self.assertEqual(gateway._compute_consensus([]), "escalate") + + +class TestPermissionErrorStatus(unittest.TestCase): + def test_maps_auth_capacity_and_policy(self) -> None: + self.assertEqual( + gateway._http_status_for_permission_error(PermissionError("missing bearer token")), + HTTPStatus.UNAUTHORIZED, + ) + self.assertEqual( + gateway._http_status_for_permission_error(PermissionError("OIDC token is expired")), + HTTPStatus.UNAUTHORIZED, + ) + self.assertEqual( + gateway._http_status_for_permission_error(PermissionError("rate limit exceeded: 10 requests per 60s")), + HTTPStatus.TOO_MANY_REQUESTS, + ) + self.assertEqual( + gateway._http_status_for_permission_error(PermissionError("too many active jobs: max 10")), + HTTPStatus.TOO_MANY_REQUESTS, + ) + self.assertEqual( + gateway._http_status_for_permission_error(PermissionError("repository is not allowlisted")), + HTTPStatus.FORBIDDEN, + ) + self.assertEqual( + gateway._http_status_for_permission_error( + PermissionError("platform_bugfix manual approval requires GitHub OIDC") + ), + HTTPStatus.FORBIDDEN, + ) + + +class TestLegacyDeployFreeze(unittest.TestCase): + def test_deploy_script_runs_gateway_and_skips_legacy_install(self) -> None: + script = Path(__file__).resolve().parents[1] / "scripts" / "deploy_codex_audit_service.sh" + text = script.read_text(encoding="utf-8") + self.assertIn("python3 -m service.ai_gateway_service", text) + self.assertNotIn('install_file "scripts/codex_audit_service.py"', text) + legacy = Path(__file__).resolve().parents[1] / "scripts" / "codex_audit_service.py" + self.assertIn("FROZEN", legacy.read_text(encoding="utf-8").splitlines()[1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_provider_scenarios.py b/tests/test_provider_scenarios.py new file mode 100644 index 00000000..ec027eef --- /dev/null +++ b/tests/test_provider_scenarios.py @@ -0,0 +1,123 @@ +"""Tests for named provider call scenarios: adaptive defaults and forced pins.""" + +from __future__ import annotations + +import unittest + +from service.contracts import MODE_REVIEW_AND_FIX, MODE_REVIEW_ONLY, PROVIDER_CODEX, PROVIDER_CURSOR +from service import provider_scenarios as scenarios + + +class ProviderScenarioTests(unittest.TestCase): + def test_adaptive_research_diagnosis_defaults(self) -> None: + kwargs = scenarios.resolve_execute_kwargs(scenarios.SCENARIO_RESEARCH_TASK_DIAGNOSIS) + self.assertEqual( + kwargs, + { + "mode": MODE_REVIEW_ONLY, + "research_stage": "drift_analysis", + "allowed_providers": [PROVIDER_CODEX], + "complexity": "medium", + }, + ) + self.assertNotIn("model", kwargs) + self.assertNotIn("reasoning_effort", kwargs) + + def test_cursor_env_only_applies_to_canary_eligible_scenarios(self) -> None: + providers = (PROVIDER_CURSOR,) + for scenario_id in scenarios.cursor_canary_scenarios(): + kwargs = scenarios.resolve_execute_kwargs(scenario_id, research_providers=providers) + self.assertEqual(kwargs["allowed_providers"], [PROVIDER_CURSOR]) + self.assertEqual(kwargs["mode"], MODE_REVIEW_ONLY) + + for scenario_id in ( + scenarios.SCENARIO_PROMOTION_PRIMARY_REVIEW, + scenarios.SCENARIO_SOXL_RSI2_CODEGEN, + scenarios.SCENARIO_GLOBAL_ETF_CODEGEN, + scenarios.SCENARIO_PLATFORM_BUGFIX, + scenarios.SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE, + ): + self.assertEqual( + scenarios.resolve_allowed_providers(scenario_id, research_providers=providers), + [PROVIDER_CODEX], + ) + + def test_forced_cursor_on_fixed_scenario_fails_closed(self) -> None: + with self.assertRaisesRegex(ValueError, "does not allow Cursor"): + scenarios.resolve_execute_kwargs( + scenarios.SCENARIO_PROMOTION_PRIMARY_REVIEW, + allowed_providers=[PROVIDER_CURSOR], + ) + + def test_forced_model_and_effort_pass_through(self) -> None: + kwargs = scenarios.resolve_execute_kwargs( + scenarios.SCENARIO_SEMANTIC_QUALITY_ACCEPTANCE, + model="gpt-5.6-terra", + reasoning_effort="medium", + ) + self.assertEqual(kwargs["model"], "gpt-5.6-terra") + self.assertEqual(kwargs["reasoning_effort"], "medium") + self.assertEqual(kwargs["allowed_providers"], [PROVIDER_CODEX]) + + def test_promotion_pins_xhigh_and_ignores_cursor_env(self) -> None: + kwargs = scenarios.resolve_execute_kwargs( + scenarios.SCENARIO_PROMOTION_PRIMARY_REVIEW, + research_providers=(PROVIDER_CODEX, PROVIDER_CURSOR), + ) + self.assertEqual(kwargs["allowed_providers"], [PROVIDER_CODEX]) + self.assertEqual(kwargs["research_stage"], "promotion_review") + self.assertEqual(kwargs["reasoning_effort"], "xhigh") + self.assertEqual(kwargs["complexity"], "high") + + def test_forced_effort_conflict_with_pin_fails(self) -> None: + with self.assertRaisesRegex(ValueError, "requires reasoning_effort=xhigh"): + scenarios.resolve_execute_kwargs( + scenarios.SCENARIO_PROMOTION_PRIMARY_REVIEW, + reasoning_effort="medium", + ) + + def test_platform_bugfix_allows_review_and_fix_but_not_cursor(self) -> None: + kwargs = scenarios.resolve_execute_kwargs( + scenarios.SCENARIO_PLATFORM_BUGFIX, + mode=MODE_REVIEW_AND_FIX, + research_providers=(PROVIDER_CURSOR,), + ) + self.assertEqual(kwargs["mode"], MODE_REVIEW_AND_FIX) + self.assertEqual(kwargs["allowed_providers"], [PROVIDER_CODEX]) + self.assertNotIn("research_stage", kwargs) + + def test_api_scenarios_are_not_execute_routes(self) -> None: + with self.assertRaisesRegex(ValueError, "not an execute scenario"): + scenarios.resolve_execute_kwargs(scenarios.SCENARIO_API_ANALYZE) + with self.assertRaisesRegex(ValueError, "API endpoint"): + scenarios.resolve_allowed_providers(scenarios.SCENARIO_API_DUAL_REVIEW) + + def test_unknown_scenario_fail_closed(self) -> None: + with self.assertRaisesRegex(ValueError, "unknown provider scenario"): + scenarios.get_scenario("trade_execution") + + def test_canary_set_is_explicit_and_small(self) -> None: + self.assertEqual( + scenarios.cursor_canary_scenarios(), + ( + scenarios.SCENARIO_RESEARCH_TASK_DIAGNOSIS, + scenarios.SCENARIO_PORTFOLIO_PROPOSAL_DIAGNOSIS, + scenarios.SCENARIO_DAILY_BRIEFING, + ), + ) + self.assertEqual( + scenarios.cursor_canary_research_stages(), + frozenset({"drift_analysis", "research_summary"}), + ) + + def test_fixed_codex_may_override_research_stage(self) -> None: + kwargs = scenarios.resolve_execute_kwargs( + scenarios.SCENARIO_RESEARCH_SUMMARY, + research_stage="optimization", + ) + self.assertEqual(kwargs["research_stage"], "optimization") + self.assertEqual(kwargs["allowed_providers"], [PROVIDER_CODEX]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_quota.py b/tests/test_quota.py index 6c4f5a88..2e98f67b 100644 --- a/tests/test_quota.py +++ b/tests/test_quota.py @@ -196,6 +196,15 @@ def test_codex_execute_does_not_consume_api_budget(self) -> None: remaining = self.manager.remaining_daily("test/repo") self.assertEqual(remaining, DEFAULT_DAILY_BUDGET_USD) + def test_cursor_execute_does_not_consume_api_budget(self) -> None: + before = self.manager.remaining_daily("test/repo") + self.manager.record_execute("test/repo", provider="cursor") + self.assertEqual(self.manager.remaining_daily("test/repo"), before) + status = self.manager.status("test/repo") + self.assertEqual(status["cursor_calls"], 1) + self.assertEqual(status["api_key_cost_usd"], 0.0) + self.assertTrue(status["cost_incomplete"]) + def test_codex_account_requires_available_unexhausted_percentages(self) -> None: def snapshot(primary, secondary=None): return {"status": "available", "rate_limits": {"primary": primary, "secondary": secondary}} diff --git a/tests/test_run_cn_index_etf_research.py b/tests/test_run_cn_index_etf_research.py index 2f2374c5..81562a75 100644 --- a/tests/test_run_cn_index_etf_research.py +++ b/tests/test_run_cn_index_etf_research.py @@ -413,8 +413,15 @@ def test_summary_callback_uses_codex_contract_and_trusted_route(): "provider": "codex", "model": "codex-test"} args, kwargs = client.execute.call_args assert "不可信 data" in args[0] and "动量+趋势+基准risk-off" in args[0] - assert kwargs == {"mode": "review_only", "research_stage": "optimization", "allowed_providers": ["codex"], - "source_repository": job.STRATEGY_REPOSITORY, "source_ref": REVISION, "timeout": 600} + assert kwargs == { + "mode": "review_only", + "research_stage": "optimization", + "allowed_providers": ["codex"], + "complexity": "low", + "source_repository": job.STRATEGY_REPOSITORY, + "source_ref": REVISION, + "timeout": 600, + } @pytest.mark.parametrize("output", [ @@ -1150,7 +1157,7 @@ def http(request, **_): else: assert job._diagnosis(runtime, drift, REVISION)()["optimization_needed"] is False if denied: - assert checked == [401] + assert checked and checked[0] in {401, 403} submit.assert_not_called() quota.record_execute.assert_not_called() else: diff --git a/tests/test_run_soxl_manual_learning.py b/tests/test_run_soxl_manual_learning.py index 144241f9..537da26e 100644 --- a/tests/test_run_soxl_manual_learning.py +++ b/tests/test_run_soxl_manual_learning.py @@ -476,7 +476,7 @@ def run_command(argv: list[str]) -> SimpleNamespace: assert len(client.calls) == 1 assert client.calls[0][1] == { - "mode": "review_only", "research_stage": "optimization", + "mode": "review_only", "research_stage": "optimization", "complexity": "medium", "allowed_providers": ["codex"], "source_repository": "QuantStrategyLab/AIAuditBridge", "source_ref": "main", "timeout": 600, } diff --git a/tests/test_subscription_execution.py b/tests/test_subscription_execution.py index 89971f07..85b1ec2a 100644 --- a/tests/test_subscription_execution.py +++ b/tests/test_subscription_execution.py @@ -209,21 +209,24 @@ def test_cursor_route_requires_fresh_account_and_explicit_spend_quality_policy() roster = {'status': 'available', 'source': 'cursor_cli_account', 'updated_at': 900, 'models': ['synthetic-model', 'unknown-new-model']} policy = {'valid_until': 2000, 'on_demand_disabled_verified': True, 'max_daily_calls': 2, - 'models': {'synthetic-model': {'quality_level': 2, 'supported_reasoning_efforts': ['high']}}} - payload = {'research_stage': 'optimization', 'mode': 'review_only'} + 'models': {'synthetic-model': {'quality_level': 1, 'supported_reasoning_efforts': ['medium']}}} + payload = {'research_stage': 'drift_analysis', 'mode': 'review_only'} result = resolve_cursor_route(payload, {}, policy=policy, roster=roster, now=1000) - assert result == {'action': 'run', 'provider': 'cursor', 'model': 'synthetic-model', 'reasoning_effort': 'high'} + assert result == {'action': 'run', 'provider': 'cursor', 'model': 'synthetic-model', 'reasoning_effort': 'medium'} for change in ({'status': 'stale'}, {'updated_at': 1001}, {'updated_at': -90000}, {'source': 'openai_api'}, {'models': ['unknown-new-model']}): assert resolve_cursor_route(payload, {}, policy=policy, roster={**roster, **change}, now=1000)['action'] == 'defer' for change in ({'on_demand_disabled_verified': False}, {'valid_until': 999}, {'max_daily_calls': 0}, {'models': {}}): assert resolve_cursor_route(payload, {}, policy={**policy, **change}, roster=roster, now=1000)['action'] == 'defer' assert resolve_cursor_route(payload, {'cursor_calls': 2}, policy=policy, roster=roster, now=1000)['action'] == 'defer' + assert resolve_cursor_route( + {'research_stage': 'optimization', 'mode': 'review_only'}, {}, policy=policy, roster=roster, now=1000, + )['reason'] == 'cursor_stage_not_canary' def test_fallback_only_after_eligible_pre_execution_codex_deferral(): quota = Mock() quota._codex_account_snapshot.return_value = None - payload = {'prompt': 'synthetic', 'mode': 'review_only', 'research_stage': 'optimization', 'allowed_providers': ['codex', 'cursor']} + payload = {'prompt': 'synthetic', 'mode': 'review_only', 'research_stage': 'drift_analysis', 'allowed_providers': ['codex', 'cursor']} with patch.dict(gateway.os.environ, {'AI_GATEWAY_CURSOR_FALLBACK_ENABLED': 'true'}, clear=True), patch.object( gateway, 'resolve_codex_research_route', return_value={'action': 'defer', 'reason': 'codex_quota_reserved', 'retry_at': 2000} ), patch.object(gateway, '_admit_cursor_execute', return_value=None) as cursor: @@ -232,9 +235,25 @@ def test_fallback_only_after_eligible_pre_execution_codex_deferral(): cursor.reset_mock() gateway._admit_codex_execute(quota, 'Synthetic/caller', {**payload, 'model': 'explicit-codex'}) gateway._admit_codex_execute(quota, 'Synthetic/caller', {**payload, 'allowed_providers': ['codex']}) + gateway._admit_codex_execute( + quota, 'Synthetic/caller', + {**payload, 'research_stage': 'optimization', 'allowed_providers': ['codex', 'cursor']}, + ) cursor.assert_not_called() +def test_cursor_rejected_for_non_canary_research_stages(): + quota = Mock() + for stage in ('optimization', 'promotion_review'): + denial = gateway._admit_codex_execute( + quota, 'Synthetic/caller', + {'prompt': 'synthetic', 'mode': 'review_only', 'research_stage': stage, 'allowed_providers': ['cursor']}, + ) + assert denial == { + 'status': 'deferred', 'error': 'cursor_stage_not_canary', 'retry_at': None, 'execution_started': False, + } + + def test_soxl_codegen_admission_allows_pinned_luna_but_defers_reserved_quota(): quota = Mock() quota._codex_account_snapshot.return_value = { @@ -332,16 +351,24 @@ def test_started_codex_failure_never_starts_cursor(): from types import SimpleNamespace job = {'job_id': 'synthetic', 'status': 'queued', 'task': 'execute', 'provider': 'codex', 'model': 'synthetic-model', 'research_stage': 'drift_analysis', 'reasoning_effort': 'medium'} + codex = SimpleNamespace(execute=Mock(return_value=SimpleNamespace(success=False, error='codex exec timed out', output=''))) + cursor = SimpleNamespace(execute=Mock()) + selected = [] + + def resolve(provider): + selected.append(provider) + return cursor if provider == 'cursor' else codex + with patch.object(gateway, '_read_job', return_value=job), patch.object(gateway, '_write_job'), patch.object( gateway, '_record_job_automation_run' ), patch.object(gateway, '_audit_log'), patch.object(gateway, 'get_health_monitor'), patch.object( gateway, '_record_platform_execution_telemetry' - ), patch.object(gateway, 'CodexAdapter') as codex, patch.object(gateway, 'CursorAdapter') as cursor: - codex.return_value.execute.return_value = SimpleNamespace(success=False, error='codex exec timed out', output='') + ), patch.object(gateway, 'resolve_execution_adapter', side_effect=resolve): gateway._run_job('synthetic', {'prompt': 'synthetic', 'allowed_providers': ['codex', 'cursor'], **job}) assert job['status'] == 'failed' - codex.return_value.execute.assert_called_once() - cursor.assert_not_called() + assert selected == ['codex'] + codex.execute.assert_called_once() + cursor.execute.assert_not_called() def test_cursor_usage_is_global_and_corrupt_store_cannot_reset_allowance(tmp_path): diff --git a/tests/test_watchdog_repair_rehearsal.py b/tests/test_watchdog_repair_rehearsal.py index f1f46b9c..c3533bdb 100644 --- a/tests/test_watchdog_repair_rehearsal.py +++ b/tests/test_watchdog_repair_rehearsal.py @@ -105,6 +105,7 @@ def execute(self, prompt: str, **kwargs: object) -> SimpleNamespace: "sandbox": "read-only", "research_stage": "drift_analysis", "allowed_providers": ["codex"], + "complexity": "medium", "source_repository": "QuantStrategyLab/AIAuditBridge", "source_ref": "main", "timeout": 600,