feat: add Super Alpha research, management and MCP workflows

backend/app/ai/contracts.py

@@ -45,6 +45,8 @@ class PageContext(Contract):
     page: Literal[
         "home",
         "alphas",
+        "superalphas",
+        "superalpha-research",
         "account",
         "datasets",
         "fields",

backend/app/alphas.py

@@ -189,6 +189,10 @@ async def upsert_alpha(db, raw: dict):
 
 def list_statement(filters):
     query = select(Alpha, Research).join(Research, Research.alpha_id == Alpha.id)
+    if filters.management_scope == "super":
+        query = query.where(Alpha.alpha_type == "SUPER")
+    elif filters.management_scope == "non_super":
+        query = query.where(or_(Alpha.alpha_type != "SUPER", Alpha.alpha_type.is_(None)))
     if filters.submission:
         query = query.where(submission_condition(filters.submission))
     if filters.submission_blocked is not None:
@@ -285,6 +289,11 @@ def summary(item: Alpha, research: Research):
     result = {k: getattr(item, k) for k in keys}
     result["failed_checks"] = failed_checks(item.checks)
     result["expression_preview"] = (item.expression or item.selection or "")[:240]
+    result["selection_preview"], result["combo_preview"] = (item.selection or "")[:240], (item.combo or "")[:240]
+    if item.alpha_type == "SUPER":
+        from .superalpha.evidence import parse_components
+        components = parse_components(item.raw.get("components", item.raw.get("selectedAlphas")))
+        result["component_count"] = len(components["components"]) if components["complete"] else None
     result["research"] = {
         k: getattr(research, k) for k in ("note", "tags", "favorite", "state", "updated_at", "version")
     }

backend/app/backtests/contracts.py

@@ -27,25 +27,47 @@ class SimulationSettings(Contract):
     maxPosition: Literal["ON", "OFF"] = "OFF"
 
 
+class SuperSimulationSettings(SimulationSettings):
+    """SUPER-only selection settings; platform metadata still determines availability."""
+
+    selectionHandling: Literal["POSITIVE", "NON_ZERO", "NON_NAN"]
+    selectionLimit: int = Field(ge=1, le=100000, strict=True)
+    componentActivation: Literal["IS", "OS"]
+
+
 class Candidate(Contract):
     client_item_id: str = Field(min_length=1, max_length=100)
-    expression: str = Field(min_length=1, max_length=20000)
-    settings: SimulationSettings
-    alpha_type: Literal["REGULAR"] = "REGULAR"
+    expression: str = Field(default="", max_length=20000)
+    selection: str | None = Field(default=None, max_length=20000)
+    combo: str | None = Field(default=None, max_length=20000)
+    settings: SuperSimulationSettings | SimulationSettings
+    alpha_type: Literal["REGULAR", "SUPER"] = "REGULAR"
 
-    @field_validator("expression")
+    @field_validator("expression", "selection", "combo")
     @classmethod
     def nonempty(cls, value):
-        value = value.strip()
-        if not value:
-            raise ValueError("表达式不能为空")
-        return value
+        return value.strip() if value is not None else None
+
+    @model_validator(mode="after")
+    def typed_input(self):
+        if self.alpha_type == "SUPER":
+            if self.expression or not self.selection or not self.combo:
+                raise ValueError("SUPER 必须提供非空 selection/combo，不能提供 regular expression")
+            if not isinstance(self.settings, SuperSimulationSettings):
+                raise ValueError("SUPER 必须提供 selectionHandling、selectionLimit、componentActivation")
+        elif not self.expression or self.selection is not None or self.combo is not None or isinstance(self.settings, SuperSimulationSettings):
+            raise ValueError("REGULAR 必须提供非空 expression，不能包含 SUPER 表达式或设置")
+        return self
 
     def platform_input(self):
+        if self.alpha_type == "SUPER":
+            return {"type": "SUPER", "selection": self.selection, "combo": self.combo,
+                    "settings": self.settings.model_dump()}
         return {"type": self.alpha_type, "regular": self.expression, "settings": self.settings.model_dump()}
 
 
 class Source(Contract):
+    research_kind: str | None = Field(default=None, max_length=50)
     kind: str = Field(default="manual", min_length=1, max_length=100)
     reference: str | None = Field(default=None, max_length=200)
     batch_id: str | None = Field(default=None, max_length=200)
@@ -54,6 +76,9 @@ class Source(Contract):
     research_id: str | None = Field(default=None, max_length=200)
     parent_run_id: str | None = Field(default=None, max_length=36)
     hypothesis: str | None = Field(default=None, max_length=2000)
+    superalpha_plan_id: str | None = Field(default=None, max_length=36)
+    superalpha_plan_version: int | None = Field(default=None, ge=1)
+    selection_snapshot_ids: list[str] = Field(default_factory=list, max_length=100)
 
 
 class SourceOutput(Source):
@@ -129,7 +154,7 @@ def fingerprint(payload: dict) -> str:
 
 def group_key(candidate: dict):
     settings = candidate["settings"]
-    return tuple(settings[k] for k in ("region", "delay", "language", "instrumentType"))
+    return (candidate.get("alpha_type", "REGULAR"), *tuple(settings[k] for k in ("region", "delay", "language", "instrumentType")))
 
 
 class ReferenceInput(Contract):
@@ -203,7 +228,10 @@ class ItemOutput(Contract):
     id: str
     client_item_id: str
     expression: str
-    settings: SimulationSettings
+    alpha_type: Literal["REGULAR", "SUPER"] = "REGULAR"
+    selection: str | None = None
+    combo: str | None = None
+    settings: SuperSimulationSettings | SimulationSettings
     attempt_id: str
     platform_status: str
     collection_status: str

backend/app/backtests/routes.py

@@ -117,13 +117,14 @@ async def runs(
     source: str | None = Query(None, max_length=100),
     reference: str | None = Query(None, max_length=200),
     research_id: str | None = Query(None, max_length=200),
+    alpha_type: Literal["REGULAR", "SUPER"] | None = None,
     q: str = Query("", max_length=200),
     sort: Literal["name", "created_at"] = "created_at",
     direction: Literal["asc", "desc"] = "desc",
 ):
     async with request.app.state.sessions() as db:
         return await Business(db).backtests.runs(
-            limit, offset, source, reference, research_id, q, sort, direction
+            limit, offset, source, reference, research_id, q, sort, direction, alpha_type
         )
 
 
@@ -195,3 +196,17 @@ async def attach_reference(attempt_id: str, body: ReferenceInput, request: Reque
 async def subset(preview_id: str, body: SubsetInput, request: Request):
     async with request.app.state.sessions.begin() as db:
         return await Business(db).backtests.subset(preview_id, body)
+
+
+@router.get("/items/{item_id}/artifact")
+async def artifact(item_id: str, request: Request, kind: Literal["snapshot", "components", "pnl"], limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    from fastapi import HTTPException
+
+    from ..research_access.contracts import Artifact
+    from ..research_access.queries import EvidenceQueries
+    from ..research_access.service import ResearchError
+    async with request.app.state.sessions() as db:
+        try:
+            return await EvidenceQueries(db).artifact(Artifact(item_id=item_id, kind=kind, limit=limit, offset=offset))
+        except ResearchError as exc:
+            raise HTTPException(404, str(exc)) from None

backend/app/backtests/runtime.py

@@ -448,7 +448,7 @@ class BacktestLane:
 
     def safe_progress(self, value):
         # Store useful protocol evidence, never arbitrary upstream diagnostics or credentials.
-        result = {k: value[k] for k in ("status", "alpha", "regular", "settings", "location") if k in value}
+        result = {k: value[k] for k in ("status", "type", "alpha", "regular", "selection", "combo", "settings", "location", "warnings") if k in value}
         message = value.get("error") or value.get("message")
         if isinstance(message, str):
             for secret in list(self.client.credentials or ()) + list(self.client.client.cookies.values()):
@@ -485,11 +485,13 @@ class BacktestLane:
             matched = [
                 i
                 for i in items
-                if i.expression == expression
+                if ((i.alpha_type == "REGULAR" and evidence.get("type", "REGULAR") == "REGULAR" and i.expression == expression)
+                    or (i.alpha_type == "SUPER" and evidence.get("type") == "SUPER"
+                        and i.selection == code(evidence.get("selection")) and i.combo == code(evidence.get("combo"))))
                 and isinstance(settings, dict)
                 and all(k in settings and settings[k] == v for k, v in i.settings.items())
             ]
-            if count == 1:
+            if count == 1 and items[0].alpha_type == "REGULAR":
                 matched = (
                     items
                     if (expression == items[0].expression or (not expression and detail is None))
@@ -499,6 +501,9 @@ class BacktestLane:
                     )
                     else []
                 )
+            if count == 1 and items[0].alpha_type == "SUPER" and detail is None:
+                # A known receipt can record progress, but saving SUPER requires full type/input evidence.
+                matched = items if not any(k in evidence for k in ("type", "selection", "combo", "settings")) else matched
             # Identical inputs within a multi-submit are intentionally not position-matched.
             if len(matched) != 1 or (matched[0].simulation_id not in (None, child)):
                 return
@@ -513,6 +518,9 @@ class BacktestLane:
                 if not await db.get(BacktestResult, item.id):
                     from datetime import datetime
 
+                    if item.alpha_type == "SUPER":
+                        from ..superalpha.evidence import save_actual_components
+                        await save_actual_components(db, item, detail, receipt["observed_at"])
                     db.add(
                         BacktestResult(
                             item_id=item.id,

backend/app/backtests/service.py

@@ -93,7 +93,7 @@ class Backtests:
 
     async def capabilities(self):
         return {
-            "alpha_types": ["REGULAR"],
+            "alpha_types": ["REGULAR", "SUPER"],
             "languages": ["FASTEXPR"],
             "instrument_types": ["EQUITY"],
             "settings_schema": Candidate.model_json_schema(),
@@ -111,6 +111,8 @@ class Backtests:
         from ..research.expressions import analyze
         if not body.preparation_refs and not body.input_ids:
             return
+        if any(c.alpha_type == "SUPER" for c in body.candidates):
+            raise HTTPException(422, "SUPER 组件快照不能使用字段数据准备集合")
         await Preparations(self.db).bind(body)
         snapshots = [await Preparations(self.db).snapshot(i) for i in body.input_ids]
         for candidate in body.candidates:
@@ -221,6 +223,8 @@ class Backtests:
                 if len(candidates) != len(selection):
                     raise HTTPException(422, "选择包含不属于当前草稿的候选")
             data = {"name": draft.name, "source": draft.source, "candidates": candidates}
+        from ..superalpha.service import validate_source
+        await validate_source(self.db, data["source"], data["candidates"])
         candidates = DraftInput.model_validate(data).model_dump(mode="json")["candidates"]
         config = await self.db.get(BacktestConfig, 1)
         groups = defaultdict(list)
@@ -253,6 +257,9 @@ class Backtests:
         for indices in groups.values():
             local_batches = []
             for index in indices:
+                if candidates[index]["alpha_type"] == "SUPER":
+                    batches.append([index])
+                    continue
                 batch = next(
                     (
                         b
@@ -352,6 +359,7 @@ class Backtests:
                         ordinal=i,
                         client_item_id=c.client_item_id,
                         expression=c.expression,
+                        alpha_type=c.alpha_type, selection=c.selection, combo=c.combo,
                         settings=c.settings.model_dump(),
                         fingerprint=fingerprint(c.platform_input()),
                     )
@@ -360,8 +368,10 @@ class Backtests:
         await self.db.flush()
         return await self.run(run.id)
 
-    async def runs(self, limit=25, offset=0, source=None, reference=None, research_id=None, q="", sort="created_at", direction="desc"):
+    async def runs(self, limit=25, offset=0, source=None, reference=None, research_id=None, q="", sort="created_at", direction="desc", alpha_type=None):
         query = select(BacktestRun)
+        if alpha_type:
+            query = query.where(BacktestRun.id.in_(select(BacktestItem.run_id).where(BacktestItem.alpha_type == alpha_type)))
         if q:
             query = query.where(BacktestRun.name.contains(q, autoescape=True))
         column = {"name": BacktestRun.name, "created_at": BacktestRun.created_at}[sort]
@@ -460,7 +470,7 @@ class Backtests:
                             for k in (
                                 "id",
                                 "client_item_id",
-                                "expression",
+                                "expression", "alpha_type", "selection", "combo",
                                 "settings",
                                 "attempt_id",
                                 "platform_status",
@@ -610,7 +620,8 @@ class Backtests:
                     source=Source.model_validate({**run.source, "parent_run_id": run.id}),
                     candidates=[
                         Candidate(
-                            client_item_id=r.client_item_id, expression=r.expression, settings=r.settings
+                            client_item_id=r.client_item_id, expression=r.expression, settings=r.settings,
+                            alpha_type=r.alpha_type, selection=r.selection, combo=r.combo
                         )
                         for r in selected
                     ],

backend/app/business.py

@@ -91,27 +91,29 @@ class Business:
             else None,
         }
 
-    async def get_alpha_facets(self):
+    async def get_alpha_facets(self, management_scope=None):
+        from .schemas import AlphaFilters
+        ids = list_statement(AlphaFilters(management_scope=management_scope)).with_only_columns(Alpha.id)
         result = {}
         for key in ("region", "universe", "alpha_type", "language", "status", "stage"):
             column = getattr(Alpha, key)
             result[key] = list(
                 (
                     await self.db.scalars(
-                        select(column).where(column.is_not(None)).distinct().order_by(column)
+                        select(column).where(column.is_not(None), Alpha.id.in_(ids)).distinct().order_by(column)
                     )
                 ).all()
             )
         result["tags"] = list(
-            (await self.db.scalars(select(ResearchTag.tag).distinct().order_by(ResearchTag.tag))).all()
+            (await self.db.scalars(select(ResearchTag.tag).where(ResearchTag.alpha_id.in_(ids)).distinct().order_by(ResearchTag.tag))).all()
         )
-        result["total"] = await self.db.scalar(select(func.count()).select_from(Alpha))
+        result["total"] = await self.db.scalar(select(func.count()).select_from(Alpha).where(Alpha.id.in_(ids)))
         result["favorites"] = await self.db.scalar(
-            select(func.count()).select_from(Research).where(Research.favorite.is_(True))
+            select(func.count()).select_from(Research).where(Research.favorite.is_(True), Research.alpha_id.in_(ids))
         )
-        result["last_sync"] = await self.db.scalar(select(func.max(Alpha.synced_at)))
+        result["last_sync"] = await self.db.scalar(select(func.max(Alpha.synced_at)).where(Alpha.id.in_(ids)))
         result["source"] = sorted(
-            {kind for kinds in (await source_kinds(self.db)).values() for kind in kinds}
+            {kind for kinds in (await source_kinds(self.db, list(await self.db.scalars(ids)))).values() for kind in kinds}
         )
         return result
 

backend/app/catalog/research_metadata.py

@@ -66,6 +66,7 @@ def setting_rows(data):
                         for key in (
                             "decay", "truncation", "pasteurization", "unitHandling",
                             "nanHandling", "language", "visualization", "maxTrade", "maxPosition",
+                            "selectionHandling", "selectionLimit", "componentActivation",
                         ):
                             definition = children.get(key)
                             if not isinstance(definition, dict):
@@ -200,11 +201,15 @@ class ResearchMetadata:
                 raise HTTPException(502, "算子分页提前结束")
         raise HTTPException(502, "算子分页超过本地限制，未发布新快照")
 
-    async def operators(self, q="", category=None, favorite=False, limit=25, offset=0):
+    async def operators(self, q="", category=None, favorite=False, limit=25, offset=0, stage=None):
         snapshot = await self.get("operators")
         notes = {r.name: r for r in await self.db.scalars(select(OperatorNote))}
         rows = []
         for item in snapshot["content"].get("items", []):
+            scopes = item.get("scope") or []
+            scopes = scopes if isinstance(scopes, list) else [scopes]
+            if stage and stage.upper() not in [str(s).upper() for s in scopes]:
+                continue
             note = notes.get(item["name"])
             if q.lower() not in json.dumps(item, ensure_ascii=False).lower() or (
                 category and item["category"] != category

backend/app/catalog/research_routes.py

@@ -18,11 +18,12 @@ async def operators(
     q: str = "",
     category: str | None = None,
     favorite: bool = False,
+    stage: str | None = None,
     limit: int = Query(25, ge=1, le=100),
     offset: int = Query(0, ge=0),
 ):
     async with request.app.state.sessions() as db:
-        return await ResearchMetadata(db).operators(q, category, favorite, limit, offset)
+        return await ResearchMetadata(db).operators(q, category, favorite, limit, offset, stage)
 
 
 @router.post("/operators/refresh")

backend/app/jobs.py

@@ -265,6 +265,10 @@ class Runner:
                     await sync_catalog(self, job_id, payload)
                 elif kind in ("full_sync", "daily_sync"):
                     await self.sync_all(job_id)
+                elif kind == "super_selection_preview":
+                    from .superalpha.jobs import run_selection
+
+                    await run_selection(self, job_id, payload)
                 elif kind == "submission_check":
                     from .submission import run_check
 
@@ -302,7 +306,7 @@ class Runner:
             await self.checkpoint(
                 job_id,
                 {
-                    "status": "waiting_connection" if waiting or (kind == "catalog_full_sync" and exc.code == "network_error") else "failed",
+                    "status": "waiting_connection" if waiting or (kind in ("catalog_full_sync", "super_selection_preview") and exc.code == "network_error") else "failed",
                     "error": str(exc),
                     "next_retry_at": None,
                 },

backend/app/main.py

@@ -6,7 +6,7 @@ import io
 import time
 from collections import defaultdict
 from contextlib import AsyncExitStack, asynccontextmanager
-from typing import Annotated
+from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request, Response
 from fastapi.exceptions import RequestValidationError
@@ -56,6 +56,7 @@ from .schemas import (
 )
 from .security import bootstrap, cipher, issue_session, require_auth, token_hash, valid_password
 from .submission import router as submission_router
+from .superalpha.routes import router as superalpha_router
 
 
 def account_output(account, client, settings):
@@ -344,9 +345,9 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
             return await Business(db).search_alphas(filters)
 
     @api.get("/alphas/facets", response_model=FacetsOutput, tags=["alphas"])
-    async def facets():
+    async def facets(management_scope: Literal["super", "non_super"] | None = None):
         async with sessions() as db:
-            return await Business(db).get_alpha_facets()
+            return await Business(db).get_alpha_facets(management_scope)
 
     @api.get(
         "/alphas/export",
@@ -487,6 +488,7 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
     app.include_router(dashboard_router)
     app.include_router(home_information_router)
     app.include_router(backtest_router)
+    app.include_router(superalpha_router)
     app.include_router(api)
     app.include_router(catalog_router)
     app.include_router(preparations_router)

backend/app/mcp_api/server.py

@@ -18,9 +18,18 @@ from ..models import MCPAudit, now
 from ..research.serialization import encode_snapshot
 from ..research_access import contracts as c
 from ..research_access.service import ResearchAccess, ResearchError
+from ..superalpha import contracts as sc
 
 # Name, schema, business method, required scope, description. No generic arbitrary HTTP tool.
 TOOLS = {
+    "search_superalpha_plans": (sc.PlanSearch, "super_plans", "research:read", "分页查找 Super Alpha 研究方案。"),
+    "get_superalpha_plan": (sc.PlanReference, "super_plan", "research:read", "读取指定方案版本或固定构造记录；不发起回测。"),
+    "save_superalpha_plan": (sc.PlanSave, "save_super_plan", "research:write", "保存调用方构造的 Selection/Combo 参数方案；更新须携带版本，支持幂等。不调用模型或回测。"),
+    "preview_superalpha_selection": (sc.SelectionPreview, "preview_super_selection", "research:refresh", "主动预览展开后的 Selection；异步返回 job_id，用 get_refresh_job 查进度、get_superalpha_selection 查组件。预览不是实际回测组件。"),
+    "get_superalpha_selection": (sc.SelectionReference, "super_selection", "research:read", "分页读取组件预览及完整性、时间、警告；缺失不自动刷新。"),
+    "build_superalpha_candidates": (sc.BuildCandidates, "build_super_candidates", "research:write", "按方案版本或内联方案进行全量展开/固定种子采样；保存固定候选及来源，不执行回测。将 candidates 与 submit_source 交给 submit_backtests；超过100项按分页读取固定记录。"),
+    "search_superalphas": (sc.SuperAlphaSearch, "super_alphas", "research:read", "分页查询本地已导入的 SUPER 成果，固定 SUPER 范围；不自动同步。"),
+    "get_superalpha": (sc.AlphaReference, "super_alpha", "research:read", "读取已导入 SUPER 的 Selection/Combo、指标、组件证据、Description 和研究来源。"),
     "get_pyramid_distribution": (c.PyramidQuery, "pyramid_distribution", "research:read", "实时读取指定 region（如 USA、GLB）和 delay（0/1）的个人 Pyramid Alpha 分布；必传 current_date（YYYY-MM-DD），自动按自然年四季度取完整起止日（如2026-09-13对应2026-07-01至2026-09-30），传给平台 startDate/endDate，不使用默认周期。按用户约定 alphaCount>=3 为 lit（已点亮），1–2 为 in_progress，0 为 unlit；每项含 category、alpha_count、距3条的 remaining。复用平台认证，未连接时先调用 authenticate_worldquant；缺失数据不当作0。不回测、不提交。"),
     "search_data_preparations": (c.PreparationSearch, "preparations", "research:read", "分页查询数据准备集合，返回固定范围、字段数及版本。研究可使用多个集合，各集合范围独立。"),
     "get_data_preparation": (c.PreparationRead, "preparation", "research:read", "按集合 ID 与版本分页预览字段、类型、描述和数据集归属。提交回测时携带 preparation_refs，由服务端核对版本并固定独立输入快照；空集合不能用于研究。"),
@@ -37,7 +46,7 @@ TOOLS = {
     "check_self_correlation": (c.SelfCorrelationCheck, "check_self_correlation", "research:refresh", "对 1–100 个已导入 Alpha 发起本地自相关检查，返回 job_id。与本地同地区已提交 Alpha 比较，排除自身；优先用缓存，缺失 PnL 自动补取。需先同步已提交 Alpha；不调用平台提交检查。用 get_refresh_job 查进度、get_self_correlation 读结果。"),
     "get_self_correlation": (c.SelfCorrelationReference, "self_correlation", "research:read", "读取指定 Alpha 最新的本地自相关缓存，包括最大相关系数、样本覆盖与 stale 状态；无缓存不自动检查，结果不等同于平台提交资格。"),
     "search_backtests": (c.History, "history", "research:read", "分页查历史候选与固定设置；candidates 按完整输入精确匹配，不推断数学等价。"),
-    "submit_backtests": (c.Submit, "submit", "backtests:execute", "执行用户已授权的固定批次，自动留痕并立即返回运行 ID。可携带 preparation_refs 选择集合，版本变化须重新读取；每项必须完整设置；重复默认拒绝，rerun 明确重跑。不需要研究资产。"),
+    "submit_backtests": (c.Submit, "submit", "backtests:execute", "执行用户已授权的 REGULAR/SUPER 固定批次，自动留痕并立即返回运行 ID；SUPER 使用 selection/combo 和专属设置，逐条模拟。可携带 preparation_refs 选择集合，版本变化须重新读取；每项必须完整设置；重复默认拒绝，rerun 明确重跑。不需要研究资产。"),
     "get_backtest": (c.RunReference, "run", "research:read", "读取真实运行进度、提交数量和可选增量事件；受理不等于成功。"),
     "get_backtest_results": (c.Results, "results", "research:read", "分页读取固定快照指标、Alpha 非通过检查及三层状态；REGULAR_SUBMISSION 单列 submission_limits，不计入 Alpha 失败统计。缺失指标不补零。"),
     "get_backtest_artifact": (c.Artifact, "artifact", "research:read", "分页读取候选脱敏快照的顶层键值或独立采集的 PnL；缺缓存不自动刷新。"),
@@ -73,8 +82,8 @@ class MCPResearchServer:
         return types.ListToolsResult(tools=[types.Tool(name=name, description=description,
             inputSchema=schema.model_json_schema(), annotations=types.ToolAnnotations(
                 readOnlyHint=scope == "research:read", destructiveHint=method == "control",
-                idempotentHint=method in {"submit", "control", "create_template"} or scope == "research:read",
-                openWorldHint=method in {"refresh", "submit", "metadata", "check_self_correlation", "check_submission", "authenticate", "pyramid_distribution"}))
+                idempotentHint=method in {"submit", "control", "create_template", "save_super_plan", "build_super_candidates"} or scope == "research:read",
+                openWorldHint=method in {"refresh", "submit", "metadata", "check_self_correlation", "check_submission", "authenticate", "pyramid_distribution", "preview_super_selection"}))
             for name, (schema, method, scope, description) in TOOLS.items()
             if scope in principal.scopes and "research:read" in principal.scopes])
 
@@ -127,7 +136,7 @@ class MCPResearchServer:
                         data = {"error": ResearchError(code, "研究操作失败；可使用原幂等键重试或查询历史", retryable=True).data}
                     db.add(MCPAudit(id=str(uuid4()), token_id=principal.token_id, tool=name,
                         request_id=fingerprint({"request_id": request_id}), input_digest=digest,
-                        business_id=data.get("backtest_run_id", data.get("job_id", data.get("template_id"))),
+                        business_id=data.get("backtest_run_id", data.get("job_id", data.get("template_id", data.get("id")))),
                         result_code=code, elapsed_ms=int((time.monotonic()-started)*1000)))
                 if not error:
                     if access.wake == "backtests":

backend/app/models.py

@@ -301,6 +301,9 @@ class BacktestItem(Base):
     client_item_id: Mapped[str] = mapped_column(String(100))
     ordinal: Mapped[int] = mapped_column(Integer)
     expression: Mapped[str] = mapped_column(Text)
+    alpha_type: Mapped[str] = mapped_column(String(20), default="REGULAR", server_default="REGULAR", index=True)
+    selection: Mapped[str | None] = mapped_column(Text)
+    combo: Mapped[str | None] = mapped_column(Text)
     settings: Mapped[dict] = mapped_column(JSON)
     fingerprint: Mapped[str] = mapped_column(String(64), index=True)
     platform_status: Mapped[str] = mapped_column(String(30), default="pending")
@@ -322,6 +325,25 @@ class BacktestResult(Base):
     complete: Mapped[bool] = mapped_column(Boolean, default=True)
 
 
+class SuperSelectionSnapshot(Base):
+    """Immutable platform component evidence; previews never replace actual components."""
+
+    __tablename__ = "super_selection_snapshots"
+    id: Mapped[str] = mapped_column(String(36), primary_key=True)
+    job_id: Mapped[str | None] = mapped_column(ForeignKey("sync_jobs.id"), unique=True)
+    item_id: Mapped[str | None] = mapped_column(ForeignKey("backtest_items.id"), unique=True)
+    source: Mapped[str] = mapped_column(String(20))
+    request: Mapped[dict] = mapped_column(JSON)
+    request_hash: Mapped[str] = mapped_column(String(64), index=True)
+    component_hash: Mapped[str | None] = mapped_column(String(64), index=True)
+    components: Mapped[list] = mapped_column(JSON, default=list)
+    raw: Mapped[dict] = mapped_column(JSON, default=dict)
+    complete: Mapped[bool] = mapped_column(Boolean, default=False)
+    total: Mapped[int | None] = mapped_column(Integer)
+    warnings: Mapped[list] = mapped_column(JSON, default=list)
+    observed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now)
+
+
 class BacktestEvent(Base):
     __tablename__ = "backtest_events"
     run_id: Mapped[str] = mapped_column(ForeignKey("backtest_runs.id"), primary_key=True)

backend/app/research/assets.py

@@ -51,7 +51,10 @@ class Assets:
         }
 
     async def save(self, body, asset_id=None, provenance=None):
+        from ..superalpha.contracts import PlanSpec
+
         schema = {
+            "superalpha_plan": PlanSpec,
             "template": TemplateSpec,
             "feature": FeatureSpec,
             "view": ViewSpec,

backend/app/research/workspace_contracts.py

@@ -11,7 +11,7 @@ from ..preparations.contracts import PreparationReference
 from ..schemas import Contract
 from .expressions import IDENTIFIER, PLACEHOLDER, normalize_template
 
-AssetKind = Literal["template", "feature", "view", "workflow"]
+AssetKind = Literal["template", "feature", "view", "workflow", "superalpha_plan"]
 
 
 class Variable(Contract):

backend/app/research_access/contracts.py

@@ -5,7 +5,7 @@ from typing import Annotated, Literal
 
 from pydantic import Field, model_validator
 
-from ..backtests.contracts import Candidate, SimulationSettings
+from ..backtests.contracts import Candidate, SimulationSettings, SuperSimulationSettings
 from ..catalog.contracts import CatalogFilters, Scope
 from ..preparations.contracts import PreparationReference
 from ..research.workspace_contracts import TemplateSpec
@@ -50,11 +50,26 @@ class CompleteSettings(SimulationSettings):
     model_config = {"json_schema_extra": {"required": list(SimulationSettings.model_fields)}}
 
 
+class CompleteSuperSettings(SuperSimulationSettings):
+    @model_validator(mode="before")
+    @classmethod
+    def complete(cls, value):
+        if isinstance(value, dict) and set(cls.model_fields) - value.keys():
+            raise ValueError("必须提供每项完整 SUPER 设置；先读取 get_research_capabilities")
+        return value
+
+    model_config = {"json_schema_extra": {"required": list(SuperSimulationSettings.model_fields)}}
+
+
 class DirectCandidate(Candidate):
-    settings: CompleteSettings
+    settings: CompleteSuperSettings | CompleteSettings
 
 
 class Provenance(Contract):
+    research_id: RunId | None = None
+    superalpha_plan_id: RunId | None = None
+    superalpha_plan_version: int | None = Field(default=None, ge=1)
+    selection_snapshot_ids: list[RunId] = Field(default_factory=list, max_length=100)
     reference: str | None = Field(default=None, max_length=200)
     batch_id: str | None = Field(default=None, max_length=200)
     hypothesis: str | None = Field(default=None, max_length=2000)
@@ -129,12 +144,18 @@ class Scopes(Contract):
 
 class SettingOptions(Page):
     kind: Literal["settings"]
+    alpha_type: Literal["REGULAR", "SUPER"] = "REGULAR"
 
 
 class Operators(Page):
     kind: Literal["operators"]
     q: str = Field(default="", max_length=300)
     category: str | None = None
+    stage: Literal["REGULAR", "SELECTION", "COMBO"] | None = None
+
+
+class SuperMetadata(Contract):
+    kind: Literal["superalpha"]
 
 
 class Availability(Contract):
@@ -144,7 +165,7 @@ class Availability(Contract):
 
 
 class Metadata(Contract):
-    query: Annotated[Scopes | SettingOptions | Operators | Availability, Field(discriminator="kind")]
+    query: Annotated[Scopes | SettingOptions | Operators | Availability | SuperMetadata, Field(discriminator="kind")]
 
 
 class CatalogRefresh(Contract):
@@ -192,6 +213,8 @@ class SubmissionCheck(Contract):
 
 
 class History(Page):
+    research_id: str | None = Field(default=None, max_length=36)
+    alpha_type: Literal["REGULAR", "SUPER"] | None = None
     source: str | None = Field(default=None, max_length=100)
     reference: str | None = Field(default=None, max_length=200)
     status: str | None = Field(default=None, max_length=30)
@@ -224,13 +247,13 @@ class Results(Page):
 
 class Artifact(Page):
     item_id: RunId
-    kind: Literal["snapshot", "pnl"]
+    kind: Literal["snapshot", "pnl", "components"]
     date_from: date | None = None
     date_to: date | None = None
 
     @model_validator(mode="after")
     def dates(self):
-        if self.kind == "snapshot" and (self.date_from or self.date_to):
+        if self.kind != "pnl" and (self.date_from or self.date_to):
             raise ValueError("日期筛选仅用于 PnL")
         if self.date_from and self.date_to and self.date_from > self.date_to:
             raise ValueError("起始日期不能晚于结束日期")

backend/app/research_access/queries.py

@@ -2,7 +2,7 @@
 
 from collections import Counter
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select
 
 from ..alphas import number, sanitize
 from ..backtests.contracts import fingerprint
@@ -54,7 +54,7 @@ def item_summary(item, result):
                             ("sharpe", "fitness", "returns", "turnover", "margin", "drawdown")}
     return encode_snapshot({
         **{k: getattr(item, k) for k in (
-            "id", "run_id", "client_item_id", "expression", "settings", "attempt_id",
+            "id", "run_id", "client_item_id", "expression", "selection", "combo", "alpha_type", "settings", "attempt_id",
             "platform_status", "collection_status", "persistence_status", "simulation_id", "alpha_id",
         )},
         "error": sanitize(item.error), "metrics": metrics,
@@ -73,10 +73,12 @@ class EvidenceQueries:
         query = select(BacktestItem, BacktestResult, BacktestRun).join(
             BacktestRun, BacktestRun.id == BacktestItem.run_id
         ).outerjoin(BacktestResult, BacktestResult.item_id == BacktestItem.id)
-        for key in ("source", "reference"):
+        for key in ("source", "reference", "research_id"):
             value = getattr(args, key)
             if value is not None:
                 query = query.where(BacktestRun.source["kind" if key == "source" else key].as_string() == value)
+        if args.alpha_type:
+            query = query.where(BacktestItem.alpha_type == args.alpha_type)
         if args.status:
             query = query.where(BacktestRun.status == args.status)
         if args.created_from:
@@ -89,7 +91,7 @@ class EvidenceQueries:
             query = query.where(BacktestItem.settings["delay"].as_integer() == args.scope.delay)
         if args.q:
             escaped = args.q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-            query = query.where(BacktestItem.expression.ilike(f"%{escaped}%", escape="\\"))
+            query = query.where(or_(*(getattr(BacktestItem, key).ilike(f"%{escaped}%", escape="\\") for key in ("expression", "selection", "combo"))))
         matches = {}
         if args.candidates:
             for c in args.candidates:
@@ -120,6 +122,9 @@ class EvidenceQueries:
         if not item:
             raise ResearchError("NOT_FOUND", "候选不存在")
         result = await self.db.get(BacktestResult, item.id)
+        if args.kind == "components":
+            from ..superalpha.evidence import actual_components
+            return await actual_components(self.db, item.id, args.limit, args.offset)
         if args.kind == "snapshot":
             # Top-level entries retain complete nested values; no hidden string/list truncation.
             entries = [{"key": k, "value": v} for k, v in sanitize(result.snapshot).items()] if result else []

backend/app/research_access/service.py

@@ -25,6 +25,7 @@ from ..research.workspace_contracts import FieldAvailabilityInput
 from ..schemas import JobInput
 from ..submission import CheckInput, correlation_allows_check, create_check_job, local_alpha, source
 from ..submission import fingerprint as submission_fingerprint
+from ..superalpha.access import SuperResearchAccess
 from ..worldquant import WqError
 from .contracts import DirectCandidate, History
 from .queries import EvidenceQueries, page
@@ -37,7 +38,7 @@ class ResearchError(Exception):
                      "retry_after": retry_after, "affected_items": affected_items or []}
 
 
-class ResearchAccess:
+class ResearchAccess(SuperResearchAccess):
     def __init__(self, db, principal, client, public_origin):
         self.db, self.principal, self.client = db, principal, client
         self.public_origin = public_origin.rstrip("/")
@@ -71,6 +72,10 @@ class ResearchAccess:
     async def capabilities(self, args):
         return {**await self.backtests.capabilities(), "max_candidates": 100,
                 "settings_schema": DirectCandidate.model_json_schema(),
+                "superalpha": {"plan_with": "save_superalpha_plan", "build_with": "build_superalpha_candidates",
+                    "preview_with": "preview_superalpha_selection", "read_selection_with": "get_superalpha_selection",
+                    "job_with": "get_refresh_job", "backtest_with": "submit_backtests", "platform_batch_size": 1,
+                    "component_evidence": "预览与实际组件分别记录，未知不能认定为同池"},
                 "confirmation": "调用者须已获本批执行授权；直接提交后返回稳定运行 ID",
                 "duplicate_policies": ["reject", "rerun"], "permissions": sorted(self.principal.scopes),
                 "metadata_only": True, "actual_platform_allowance": None,
@@ -159,15 +164,23 @@ class ResearchAccess:
     async def metadata(self, args):
         q = args.query
         metadata = ResearchMetadata(self.db)
+        if q.kind == "superalpha":
+            from ..superalpha.metadata import metadata as super_metadata
+            return await super_metadata(self.db)
         if q.kind == "scopes":
             return {"source": "worldquant_platform", **await platform_options(self.client)}
         if q.kind == "operators":
-            data = await metadata.operators(q.q, q.category, limit=q.limit, offset=q.offset)
+            data = await metadata.operators(q.q, q.category, limit=q.limit, offset=q.offset, stage=q.stage)
             return {**data, "status": "available" if data["fetched_at"] else "not_cached",
                     "has_more": q.offset + len(data["items"]) < data["total"]}
         if q.kind == "settings":
             data = await metadata.get("settings")
             items = data["content"].get("items", [])
+            special = {"selectionHandling", "selectionLimit", "componentActivation"}
+            if q.alpha_type == "REGULAR":
+                items = [{**r, "fields": {k: v for k, v in r.get("fields", {}).items() if k not in special}} for r in items]
+            else:
+                items = [{**r, "super_settings_completeness": "cached" if special <= r.get("fields", {}).keys() else "unknown"} for r in items]
             return {"status": "available" if data["fetched_at"] else "not_cached",
                     "fetched_at": data["fetched_at"], **page(items[q.offset:q.offset+q.limit], len(items), q.limit, q.offset)}
         data = await metadata.get(availability_key(q.field_id, q.scope))
@@ -201,14 +214,16 @@ class ResearchAccess:
 
     async def refresh_job(self, args):
         job = await self.db.get(Job, args.job_id)
-        if not job or job.kind not in {"catalog_sync", "field_sync", "pnl_refresh", "self_correlation", "submission_check"}:
+        if not job or job.kind not in {"catalog_sync", "field_sync", "pnl_refresh", "self_correlation", "submission_check", "super_selection_preview"}:
             raise ResearchError("NOT_FOUND", "研究刷新任务不存在")
         result = await self.business.get_job_status(args.job_id)
         query = select(JobItem).where(JobItem.job_id == job.id, JobItem.error.is_not(None))
         total = await self.db.scalar(select(func.count()).select_from(query.subquery()))
         errors = list(await self.db.scalars(query.order_by(JobItem.alpha_id).limit(args.limit).offset(args.offset)))
         result.pop("errors", None)
-        return {**result, "job_id": job.id, "artifact_reference": job.payload,
+        artifact = ({"job_id": job.id, "snapshot_id": job.checkpoint.get("snapshot_id"), "read_with": "get_superalpha_selection"}
+                    if job.kind == "super_selection_preview" else job.payload)
+        return {**result, "job_id": job.id, "artifact_reference": artifact,
                 "errors": page([{"alpha_id": e.alpha_id, "error": e.error} for e in errors], total, args.limit, args.offset)}
 
     async def check_self_correlation(self, args):
@@ -306,7 +321,11 @@ class ResearchAccess:
                 invalid.append(c.client_item_id)
         if invalid:
             raise ResearchError("UNSUPPORTED_SETTINGS", "已缓存平台设置不支持这些组合；可显式刷新后重试", affected_items=invalid)
-        return {"settings_validation": "cached", "fetched_at": snapshot["fetched_at"], "field_validation": "unknown"}
+        result = {"settings_validation": "cached", "fetched_at": snapshot["fetched_at"], "field_validation": "unknown"}
+        if any(c.alpha_type == "SUPER" for c in candidates):
+            from ..superalpha.settings import validate_settings
+            result["super_settings_validation"] = await validate_settings(self.db, [c.settings for c in candidates if c.alpha_type == "SUPER"])
+        return result
 
     async def submit(self, args):
         previous, digest = await self.previous("submit_backtests", args)

backend/app/schemas.py

@@ -69,6 +69,7 @@ class PreferencesInput(Contract):
 
 
 class AlphaFilters(Contract):
+    management_scope: Literal["super", "non_super"] | None = None
     local_correlation_status: Literal["not_cached", "stale", "low", "high", "partial", "insufficient_data"] | None = None
     local_correlation_min: float | None = Field(default=None, ge=-1, le=1)
     local_correlation_max: float | None = Field(default=None, ge=-1, le=1)
@@ -237,6 +238,9 @@ class AlphaSummary(BaseModel):
     id: str
     name: str | None
     expression_preview: str
+    selection_preview: str = ""
+    combo_preview: str = ""
+    component_count: int | None = None
     alpha_type: str | None
     language: str | None
     stage: str | None

backend/app/superalpha/__init__.py

@@ -0,0 +1 @@
+"""Super Alpha construction and immutable component evidence over shared execution."""

backend/app/superalpha/access.py

@@ -0,0 +1,43 @@
+"""MCP adapter using the same SUPER operations and records as the web editor."""
+
+from ..research.assets import Assets
+from .evidence import read_selection
+from .service import SuperResearch
+
+
+class SuperResearchAccess:
+    async def super_plans(self, args):
+        return await Assets(self.db).list("superalpha_plan", args.q, args.limit, args.offset)
+
+    async def super_plan(self, args):
+        if args.experiment_id:
+            return await SuperResearch(self.db).experiment(args.experiment_id, args.limit, args.offset)
+        return await Assets(self.db).get(args.plan_id, args.version, "superalpha_plan")
+
+    async def save_super_plan(self, args):
+        result = await SuperResearch(self.db).save(args)
+        return {**result, "web_url": f"{self.public_origin}/#superalpha-research?plan_id={result['id']}"}
+
+    async def preview_super_selection(self, args):
+        result = await SuperResearch(self.db).selection_job(args)
+        self.wake = "jobs"
+        return result
+
+    async def super_selection(self, args):
+        return await read_selection(self.db, args)
+
+    async def build_super_candidates(self, args):
+        result = await SuperResearch(self.db).build(args)
+        return {**result, "web_url": f"{self.public_origin}/#superalpha-research?experiment_id={result['id']}",
+                "submit_with": "submit_backtests", "starts_backtests": False,
+                "submit_source": {k: result["source"][k] for k in ("research_id", "superalpha_plan_id", "superalpha_plan_version", "selection_snapshot_ids", "reference", "hypothesis")},
+                "paging": "完整候选可用 get_superalpha_plan 的 experiment_id 读取"}
+
+    async def super_alphas(self, args):
+        args.filters.management_scope = "super"
+        args.filters.alpha_type = "SUPER"
+        return await self.business.search_alphas(args.filters)
+
+    async def super_alpha(self, args):
+        result = await SuperResearch(self.db).alpha(args.alpha_id)
+        return {**result, "web_url": f"{self.public_origin}/#superalphas?alpha_id={args.alpha_id}"}

backend/app/superalpha/contracts.py

@@ -0,0 +1,140 @@
+"""Bounded SUPER authoring inputs, independent from regular data-field preparation."""
+
+from typing import Literal
+
+from pydantic import Field, field_validator, model_validator
+
+from ..backtests.contracts import SuperSimulationSettings
+from ..research.expressions import IDENTIFIER, PLACEHOLDER
+from ..research.workspace_contracts import Variable
+from ..schemas import AlphaFilters, Contract
+
+
+class PlanSpec(Contract):
+    name: str = Field(min_length=1, max_length=200)
+    hypothesis: str = Field(min_length=1, max_length=2000)
+    selection: str = Field(min_length=1, max_length=20000)
+    combo: str = Field(min_length=1, max_length=20000)
+    variables: dict[str, Variable] = Field(default_factory=dict, max_length=50)
+    settings: SuperSimulationSettings
+    setting_variants: dict[str, list[str | int | float | bool]] = Field(default_factory=dict, max_length=20)
+    include_baseline: bool = False
+    reference: str = Field(default="", max_length=200)
+    parent_plan_id: str | None = Field(default=None, max_length=36)
+    parent_plan_version: int | None = Field(default=None, ge=1)
+    parent_alpha_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,100}$")
+    parent_experiment_id: str | None = Field(default=None, max_length=36)
+
+    @field_validator("name", "hypothesis", "selection", "combo")
+    @classmethod
+    def text(cls, value):
+        if not value.strip():
+            raise ValueError("内容不能为空")
+        return value.strip()
+
+    @model_validator(mode="after")
+    def bindings(self):
+        text = self.selection + "\n" + self.combo
+        if set(PLACEHOLDER.findall(text)) != set(self.variables):
+            raise ValueError("Selection/Combo 占位符必须与变量逐一对应")
+        if any(not IDENTIFIER.fullmatch(k) or v.kind == "field" for k, v in self.variables.items()):
+            raise ValueError("SUPER 变量须使用合法名称，不能使用 REGULAR 数据字段绑定")
+        if "{" in PLACEHOLDER.sub("", text) or "}" in PLACEHOLDER.sub("", text):
+            raise ValueError("占位符格式错误")
+        for key, values in self.setting_variants.items():
+            if key not in SuperSimulationSettings.model_fields or not 1 <= len(values) <= 100:
+                raise ValueError("设置变量必须为已支持设置，每项 1–100 个候选值")
+            for value in values:
+                SuperSimulationSettings.model_validate({**self.settings.model_dump(), key: value})
+        if bool(self.parent_plan_id) != bool(self.parent_plan_version):
+            raise ValueError("父方案必须同时指定 ID 和版本")
+        return self
+
+
+class PlanSearch(Contract):
+    q: str = Field(default="", max_length=200)
+    limit: int = Field(default=25, ge=1, le=100)
+    offset: int = Field(default=0, ge=0)
+
+
+class PlanReference(PlanSearch):
+    plan_id: str | None = Field(default=None, min_length=1, max_length=36)
+    experiment_id: str | None = Field(default=None, min_length=1, max_length=36)
+    version: int | None = Field(default=None, ge=1)
+
+    @model_validator(mode="after")
+    def one_reference(self):
+        if bool(self.plan_id) == bool(self.experiment_id) or (self.version and not self.plan_id):
+            raise ValueError("提供 plan_id 或 experiment_id 之一；version 仅用于方案")
+        return self
+
+
+class PlanSave(Contract):
+    plan: PlanSpec
+    plan_id: str | None = Field(default=None, max_length=36)
+    version: int | None = Field(default=None, ge=1)
+    idempotency_key: str = Field(min_length=1, max_length=100)
+
+    @model_validator(mode="after")
+    def reference(self):
+        if bool(self.plan_id) != bool(self.version):
+            raise ValueError("更新须同时提供方案 ID 与当前版本")
+        return self
+
+
+class SelectionPreview(Contract):
+    plan_id: str | None = Field(default=None, max_length=36)
+    version: int | None = Field(default=None, ge=1)
+    selection: str = Field(min_length=1, max_length=20000)
+    settings: SuperSimulationSettings
+
+    @field_validator("selection")
+    @classmethod
+    def concrete(cls, value):
+        if not value.strip() or "{" in value or "}" in value:
+            raise ValueError("预览须提供展开后的非空 Selection")
+        return value.strip()
+
+    def platform_query(self):
+        return {"selection": self.selection, **self.settings.model_dump(include={
+            "instrumentType", "region", "delay", "selectionLimit", "selectionHandling"})}
+
+
+class SelectionReference(PlanSearch):
+    snapshot_id: str | None = Field(default=None, max_length=36)
+    job_id: str | None = Field(default=None, max_length=36)
+
+    @model_validator(mode="after")
+    def one(self):
+        if bool(self.snapshot_id) == bool(self.job_id):
+            raise ValueError("提供 snapshot_id 或 job_id 之一")
+        return self
+
+
+class BuildCandidates(Contract):
+    plan: PlanSpec | None = None
+    plan_id: str | None = Field(default=None, max_length=36)
+    version: int | None = Field(default=None, ge=1)
+    mode: Literal["all", "random"] = "all"
+    limit: int = Field(default=100, ge=1, le=10000)
+    seed: int = Field(default=0, ge=0, le=2147483647)
+    selection_snapshot_ids: list[str] = Field(default_factory=list, max_length=100)
+    idempotency_key: str = Field(min_length=1, max_length=100)
+
+    @model_validator(mode="after")
+    def one(self):
+        if bool(self.plan) == bool(self.plan_id) or bool(self.plan_id) != bool(self.version):
+            raise ValueError("提供内联方案或方案 ID/版本之一")
+        return self
+
+
+class SuperAlphaSearch(Contract):
+    filters: AlphaFilters = Field(default_factory=AlphaFilters)
+
+
+class AlphaReference(Contract):
+    alpha_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,100}$")
+
+
+class ExperimentPreview(Contract):
+    candidate_ids: list[str] = Field(min_length=1, max_length=10000)

backend/app/superalpha/evidence.py

@@ -0,0 +1,97 @@
+"""Parse only explicit component evidence; never infer actual members from a preview."""
+
+import re
+from datetime import datetime
+from uuid import uuid4
+
+from fastapi import HTTPException
+from sqlalchemy import select
+
+from ..alphas import sanitize
+from ..backtests.contracts import fingerprint
+from ..models import SuperSelectionSnapshot
+from ..research.serialization import encode_snapshot
+
+
+def parse_components(raw):
+    """Return normalized rows and completeness; count/duplicate/next ambiguity stays unknown."""
+    warnings = []
+    if isinstance(raw, dict):
+        supplied = raw.get("warnings", [])
+        warnings.extend(supplied if isinstance(supplied, list) else [supplied])
+        rows = raw.get("results", raw.get("alphas", raw.get("components")))
+        total = raw.get("count", raw.get("total"))
+        complete_hint = raw.get("complete") is True
+        next_page = raw.get("next")
+    else:
+        rows, total, complete_hint, next_page = raw, None, False, None
+    invalid_total = total is not None and (type(total) is not int or total < 0)
+    total = total if type(total) is int and total >= 0 else None
+    valid_shape = isinstance(rows, list)
+    items, seen, malformed = [], set(), False
+    for row in rows if valid_shape else []:
+        entry = {"id": row} if isinstance(row, str) else row
+        if not isinstance(entry, dict):
+            malformed = True
+            continue
+        alpha_id = entry.get("id", entry.get("alpha", entry.get("alphaId")))
+        if not isinstance(alpha_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", alpha_id) or alpha_id in seen:
+            malformed = True
+            continue
+        seen.add(alpha_id)
+        items.append({**sanitize(entry), "id": alpha_id})
+    complete = valid_shape and not malformed and not invalid_total and not next_page and (
+        (total is not None and total == len(items)) or (total is None and complete_hint))
+    if not complete:
+        warnings.append("组件列表未核实完整性；不生成完整组件指纹，不用于同池结论")
+    return {"components": items, "total": total, "complete": complete,
+            "component_hash": fingerprint({"alpha_ids": sorted(seen)}) if complete else None,
+            "warnings": sanitize(warnings)}
+
+
+def snapshot_output(row, limit=25, offset=0, q=""):
+    items = [item for item in row.components if not q or q.lower() in str(item).lower()]
+    return encode_snapshot({"snapshot_id": row.id, "job_id": row.job_id, "item_id": row.item_id,
+        "source": row.source, "request": row.request, "request_hash": row.request_hash,
+        "component_hash": row.component_hash, "complete": row.complete, "reported_total": row.total,
+        "observed_at": row.observed_at, "warnings": row.warnings,
+        "status": "available" if row.complete else "unknown", "total": len(items),
+        "limit": limit, "offset": offset, "has_more": offset + limit < len(items),
+        "items": items[offset:offset + limit]})
+
+
+async def read_selection(db, args):
+    query = select(SuperSelectionSnapshot)
+    query = query.where(SuperSelectionSnapshot.id == args.snapshot_id) if args.snapshot_id else query.where(
+        SuperSelectionSnapshot.job_id == args.job_id)
+    row = await db.scalar(query)
+    if not row:
+        if args.job_id:
+            from ..models import Job
+            job = await db.get(Job, args.job_id)
+            if not job or job.kind != "super_selection_preview":
+                raise HTTPException(404, "组件预览任务不存在")
+            return {"status": job.status, "snapshot_id": None, "job_id": job.id, "items": [],
+                    "total": 0, "complete": False, "error": job.error, "observed_at": None}
+        raise HTTPException(404, "组件快照不存在")
+    return snapshot_output(row, args.limit, args.offset, args.q)
+
+
+async def save_actual_components(db, item, detail, observed_at):
+    raw = detail.get("components", detail.get("selectedAlphas"))
+    if raw is None and isinstance(detail.get("selection"), dict):
+        selection = detail["selection"]
+        if isinstance(selection.get("alphas"), list):
+            raw = {"alphas": selection["alphas"], "count": selection.get("count")}
+    request = {"type": "SUPER", "selection": item.selection, "combo": item.combo, "settings": item.settings}
+    parsed = parse_components(raw)
+    db.add(SuperSelectionSnapshot(id=str(uuid4()), item_id=item.id, source="actual", request=request,
+        request_hash=fingerprint(request), raw=sanitize(raw) if isinstance(raw, (dict, list)) else {},
+        observed_at=datetime.fromisoformat(observed_at), **parsed))
+
+
+async def actual_components(db, item_id, limit=25, offset=0):
+    row = await db.scalar(select(SuperSelectionSnapshot).where(SuperSelectionSnapshot.item_id == item_id))
+    return snapshot_output(row, limit, offset) if row else {
+        "status": "unknown", "complete": False, "source": "actual", "items": [], "total": 0,
+        "component_hash": None, "observed_at": None, "warnings": ["平台实际组件尚未核实"]}

backend/app/superalpha/jobs.py

@@ -0,0 +1,31 @@
+"""Selection previews run on the existing durable job runner, outside request transactions."""
+
+import asyncio
+from uuid import uuid4
+
+from sqlalchemy import select
+
+from ..alphas import sanitize
+from ..backtests.contracts import fingerprint
+from ..models import Job, SuperSelectionSnapshot, now
+from .contracts import SelectionPreview
+from .evidence import parse_components
+
+
+async def run_selection(runner, job_id, payload):
+    async with runner.sessions() as db:
+        existing = await db.scalar(select(SuperSelectionSnapshot).where(SuperSelectionSnapshot.job_id == job_id))
+        if existing:
+            return  # Restart after snapshot commit must not replace the original observation.
+    request = SelectionPreview.model_validate(payload)
+    raw = await runner.client.run_super_selection(request.platform_query())
+    parsed = parse_components(raw)
+    async with runner.sessions.begin() as db:
+        job = await db.get(Job, job_id)
+        if job.cancel_requested:
+            raise asyncio.CancelledError()
+        snapshot = SuperSelectionSnapshot(id=str(uuid4()), job_id=job_id, source="preview", request=payload,
+            request_hash=fingerprint(request.platform_query()), raw=sanitize(raw), **parsed)
+        db.add(snapshot)
+        job.processed, job.total, job.updated_at = 1, 1, now()
+        job.checkpoint = {"snapshot_id": snapshot.id, "complete": parsed["complete"]}

backend/app/superalpha/metadata.py

@@ -0,0 +1,27 @@
+"""Separate Alpha selection properties from stock data fields; availability remains evidence based."""
+
+from ..backtests.contracts import SuperSimulationSettings
+from ..catalog.research_metadata import ResearchMetadata
+
+
+async def metadata(db):
+    settings = await ResearchMetadata(db).get("settings")
+    return {"settings_schema": SuperSimulationSettings.model_json_schema(), "settings_snapshot": settings,
+        "selection_properties": [{"name": name, "description": description} for name, description in (
+            ("category", "用户设置的 Alpha 类别"), ("color", "用户设置的颜色"),
+            ("datasets", "组件使用的数据集集合，可配合 in()"), ("datafields", "组件使用的数据字段集合"),
+            ("datacategories", "组件使用的数据类别集合"), ("dataset_count", "不同数据集数量"),
+            ("datafield_count", "不同数据字段数量"), ("datacategory_count", "不同数据类别数量"),
+            ("decay", "组件的衰减设置"), ("favorite", "平台收藏状态"), ("name", "组件名称，按完整名称匹配"),
+            ("neutralization", "组件的中性化设置"), ("operator_count", "组件表达式算子数量"),
+            ("long_count", "IS 平均多头股票数量"), ("short_count", "IS 平均空头股票数量"),
+            ("tags", "组件的自定义标签集合"), ("truncation", "组件截断设置"),
+            ("turnover", "组件 IS 换手率"), ("universe", "组件股票池名称"),
+            ("self_correlation", "组件自相关属性"), ("prod_correlation", "组件生产相关性属性"),
+            ("os_start_date", "组件样本外起始日期，YYYY-MM-DD 字符串"),
+            ("classifications", "组件分类集合"), ("competitions", "组件关联比赛集合"))],
+        "property_source": "BRAIN Selection Expression 文档快照（2025-10-16）；属性列表非账户实时授权清单，具体可用性以平台响应为准",
+        "combo_input": "alpha 表示选中的组件；Combo 返回每日每个组件的权重，常量 1 可作为等权基线",
+        "selection_object": "平台可供选择的已提交 ACTIVE Alpha；本地列表不等同于平台完整组件池",
+        "operator_query": {"kind": "operators", "stage": "SELECTION"},
+        "validation": "结构校验与平台执行分开；缺少适用范围的算子保持未知"}

backend/app/superalpha/routes.py

@@ -0,0 +1,135 @@
+"""Authenticated SUPER authoring endpoints; construction never starts a simulation."""
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from pydantic import ValidationError
+from sqlalchemy import func, select
+
+from ..models import SuperSelectionSnapshot
+from ..research.assets import Assets
+from ..security import require_auth
+from .contracts import BuildCandidates, ExperimentPreview, PlanSave, SelectionPreview, SelectionReference
+from .evidence import read_selection, snapshot_output
+from .metadata import metadata
+from .service import SuperResearch
+
+router = APIRouter(prefix="/api/v1/superalpha", tags=["superalpha"], dependencies=[Depends(require_auth)])
+
+
+@router.get("/metadata")
+async def get_metadata(request: Request):
+    async with request.app.state.sessions() as db:
+        return await metadata(db)
+
+
+@router.get("/plans")
+async def plans(request: Request, q: str = "", limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    async with request.app.state.sessions() as db:
+        return await Assets(db).list("superalpha_plan", q, limit, offset)
+
+
+@router.post("/plans")
+async def save_plan(body: PlanSave, request: Request):
+    async with request.app.state.sessions.begin() as db:
+        return await SuperResearch(db).save(body)
+
+
+@router.get("/plans/{plan_id}")
+async def plan(plan_id: str, request: Request, version: int | None = Query(None, ge=1)):
+    async with request.app.state.sessions() as db:
+        return await Assets(db).get(plan_id, version, "superalpha_plan")
+
+
+@router.get("/plans/{plan_id}/versions")
+async def versions(plan_id: str, request: Request):
+    async with request.app.state.sessions() as db:
+        await Assets(db).get(plan_id, expected_kind="superalpha_plan")
+        return await Assets(db).versions(plan_id)
+
+
+@router.delete("/plans/{plan_id}")
+async def archive(plan_id: str, request: Request, version: int = Query(..., ge=1)):
+    async with request.app.state.sessions.begin() as db:
+        await Assets(db).get(plan_id, expected_kind="superalpha_plan")
+        return await Assets(db).archive(plan_id, version)
+
+
+@router.post("/selections", status_code=202)
+async def preview_selection(body: SelectionPreview, request: Request):
+    async with request.app.state.sessions.begin() as db:
+        result = await SuperResearch(db).selection_job(body)
+    request.app.state.runner.wake.set()
+    return result
+
+
+@router.get("/selections")
+async def selection(request: Request, snapshot_id: str | None = None, job_id: str | None = None,
+                    q: str = "", limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    try:
+        args = SelectionReference(snapshot_id=snapshot_id, job_id=job_id, q=q, limit=limit, offset=offset)
+    except ValidationError as exc:
+        raise HTTPException(422, str(exc)) from None
+    async with request.app.state.sessions() as db:
+        return await read_selection(db, args)
+
+
+@router.post("/candidates", status_code=201)
+async def build(body: BuildCandidates, request: Request):
+    async with request.app.state.sessions.begin() as db:
+        return await SuperResearch(db).build(body)
+
+
+@router.get("/experiments")
+async def experiments(request: Request, plan_id: str | None = None,
+                      limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    async with request.app.state.sessions() as db:
+        return await SuperResearch(db).experiments(plan_id, limit, offset)
+
+
+@router.get("/experiments/{experiment_id}")
+async def experiment(experiment_id: str, request: Request, limit: int = Query(100, ge=1, le=100), offset: int = Query(0, ge=0)):
+    async with request.app.state.sessions() as db:
+        return await SuperResearch(db).experiment(experiment_id, limit, offset)
+
+
+@router.post("/experiments/{experiment_id}/preview")
+async def preview(experiment_id: str, body: ExperimentPreview, request: Request):
+    async with request.app.state.sessions.begin() as db:
+        return await SuperResearch(db).preview(experiment_id, body.candidate_ids)
+
+
+@router.get("/alphas/{alpha_id}")
+async def alpha(alpha_id: str, request: Request):
+    async with request.app.state.sessions() as db:
+        return await SuperResearch(db).alpha(alpha_id)
+
+
+@router.get("/selection-history")
+async def selection_history(request: Request, plan_id: str, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    async with request.app.state.sessions() as db:
+        query = select(SuperSelectionSnapshot).where(SuperSelectionSnapshot.source == "preview", SuperSelectionSnapshot.request["plan_id"].as_string() == plan_id)
+        total = await db.scalar(select(func.count()).select_from(query.subquery()))
+        rows = await db.scalars(query.order_by(SuperSelectionSnapshot.observed_at.desc(), SuperSelectionSnapshot.id).limit(limit).offset(offset))
+        return {"items": [snapshot_output(row, 0) for row in rows], "total": total, "limit": limit, "offset": offset}
+
+
+@router.get("/experiments/{experiment_id}/results")
+async def experiment_results(experiment_id: str, request: Request, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    from ..models import Pnl
+    from ..research.serialization import encode_snapshot
+    from ..research_access.contracts import History
+    from ..research_access.queries import EvidenceQueries
+    from .evidence import actual_components
+    async with request.app.state.sessions() as db:
+        await SuperResearch(db).experiment(experiment_id, 1)
+        result = await EvidenceQueries(db).history(History(research_id=experiment_id, alpha_type="SUPER", limit=limit, offset=offset))
+        for item in result["items"]:
+            item["components"] = await actual_components(db, item["id"], 0)
+            pnl = await db.get(Pnl, item["alpha_id"]) if item["alpha_id"] else None
+            item["pnl_fetched_at"] = pnl.fetched_at if pnl else None
+        return encode_snapshot(result)
+
+
+@router.get("/alphas/{alpha_id}/components")
+async def alpha_components(alpha_id: str, request: Request, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
+    async with request.app.state.sessions() as db:
+        return (await SuperResearch(db).alpha(alpha_id, limit, offset))["components"]

backend/app/superalpha/service.py

@@ -0,0 +1,227 @@
+"""Versioned SUPER plans and deterministic candidate construction; never executes simulations."""
+
+import math
+import random
+from uuid import uuid4
+
+from fastapi import HTTPException
+from sqlalchemy import func, select
+
+from ..alphas import sanitize
+from ..backtests.contracts import Candidate, DraftInput, PreviewInput, Source, fingerprint
+from ..backtests.service import Backtests
+from ..models import Account, Alpha, Job, ResearchExperiment, ResearchRequest, SuperSelectionSnapshot, now
+from ..research.assets import Assets
+from ..research.expressions import PLACEHOLDER
+from ..research.serialization import encode_snapshot
+from ..research.workspace_contracts import AssetWrite
+from .contracts import PlanSpec, SelectionPreview
+from .evidence import actual_components, parse_components
+from .settings import validate_settings
+
+
+async def validate_source(db, source, candidates):
+    """Verify server-owned provenance references without making assets mandatory for direct execution."""
+    if (source.get("superalpha_plan_id") or source.get("selection_snapshot_ids") or source.get("kind") == "superalpha") and any(c.get("alpha_type", "REGULAR") != "SUPER" for c in candidates):
+        raise HTTPException(422, "Super Alpha 方案或组件来源只能关联 SUPER 候选")
+    await validate_settings(db, [c["settings"] for c in candidates if c.get("alpha_type") == "SUPER"])
+    experiment = None
+    if bool(source.get("superalpha_plan_id")) != bool(source.get("superalpha_plan_version")):
+        raise HTTPException(422, "方案引用须同时指定 ID 和版本")
+    if source.get("superalpha_plan_id"):
+        if not source.get("superalpha_plan_version"):
+            raise HTTPException(422, "方案引用须指定版本")
+        await Assets(db).get(source["superalpha_plan_id"], source["superalpha_plan_version"], "superalpha_plan")
+    if source.get("research_id") and (source.get("kind") == "superalpha" or source.get("superalpha_plan_id") or any(c.get("alpha_type") == "SUPER" for c in candidates)):
+        experiment = await db.get(ResearchExperiment, source["research_id"])
+        if not experiment or experiment.kind != "superalpha":
+            raise HTTPException(404, "SUPER 候选构造记录不存在")
+        source["research_kind"] = "superalpha"
+        expected = {c["client_item_id"]: fingerprint(Candidate.model_validate(c).platform_input()) for c in experiment.candidates}
+        for value in candidates:
+            c = Candidate.model_validate(value)
+            if expected.get(c.client_item_id) != fingerprint(c.platform_input()):
+                raise HTTPException(409, "候选与引用的固定构造记录不一致")
+        ref = experiment.evidence.get("plan_reference", {})
+        if source.get("superalpha_plan_id") and ref != {
+            "id": source["superalpha_plan_id"], "version": source["superalpha_plan_version"]}:
+            raise HTTPException(409, "方案版本与构造来源不一致")
+    for snapshot_id in source.get("selection_snapshot_ids", []):
+        row = await db.get(SuperSelectionSnapshot, snapshot_id)
+        if not row or row.source != "preview":
+            raise HTTPException(404, "Selection 预览快照不存在")
+        query = SelectionPreview.model_validate(row.request).platform_query()
+        if not any(Candidate.model_validate(c).alpha_type == "SUPER" and SelectionPreview(
+            selection=c["selection"], settings=c["settings"]).platform_query() == query for c in (experiment.candidates if experiment else candidates)):
+            raise HTTPException(409, "组件预览与候选 Selection/范围不匹配")
+
+
+class SuperResearch:
+    def __init__(self, db):
+        self.db = db
+        self.assets = Assets(db)
+
+    async def previous(self, operation, args):
+        account = await self.db.scalar(select(Account).where(Account.id == 1).with_for_update())
+        if not account:
+            raise HTTPException(409, "工作空间未初始化")
+        digest = fingerprint(args.model_dump(mode="json", exclude={"idempotency_key"}))
+        previous = await self.db.scalar(select(ResearchRequest).where(ResearchRequest.account_id == 1,
+            ResearchRequest.operation == operation, ResearchRequest.idempotency_key == args.idempotency_key))
+        if previous and previous.digest != digest:
+            raise HTTPException(409, "幂等键已用于不同内容")
+        return previous, digest
+
+    async def remember(self, operation, args, digest, result, business_id):
+        result = encode_snapshot(result)
+        result["_meta"] = {"schema_version": 1, "observed_at": now().isoformat(), "source": "system"}
+        self.db.add(ResearchRequest(id=str(uuid4()), account_id=1, operation=operation,
+            idempotency_key=args.idempotency_key, digest=digest, business_id=business_id, response=result))
+        await self.db.flush()
+        return result
+
+    async def provenance(self, plan):
+        result = {"reference": plan.reference}
+        if plan.parent_plan_id:
+            parent = await self.assets.get(plan.parent_plan_id, plan.parent_plan_version, "superalpha_plan")
+            result["parent_plan"] = {k: parent[k] for k in ("id", "version", "name")}
+        if plan.parent_alpha_id:
+            alpha = await self.db.get(Alpha, plan.parent_alpha_id)
+            if not alpha or alpha.alpha_type != "SUPER":
+                raise HTTPException(404, "父 SUPER Alpha 尚未导入")
+            result["parent_alpha"] = {"id": alpha.id, "snapshot": sanitize(alpha.raw), "observed_at": alpha.synced_at}
+        if plan.parent_experiment_id:
+            parent = await self.experiment(plan.parent_experiment_id)
+            result["parent_experiment"] = {"id": parent["id"], "created_at": parent["created_at"]}
+        return encode_snapshot(result)
+
+    async def save(self, args):
+        previous, digest = await self.previous("save_superalpha_plan", args)
+        if previous:
+            return previous.response
+        await validate_settings(self.db, [args.plan.settings])
+        result = await self.assets.save(AssetWrite(kind="superalpha_plan", content=args.plan.model_dump(mode="json"),
+            version=args.version), args.plan_id, await self.provenance(args.plan))
+        return await self.remember("save_superalpha_plan", args, digest, result, result["id"])
+
+    async def build(self, args):
+        previous, digest = await self.previous("build_superalpha_candidates", args)
+        if previous:
+            return previous.response
+        asset = await self.assets.get(args.plan_id, args.version, "superalpha_plan") if args.plan_id else None
+        plan = PlanSpec.model_validate(asset["content"]) if asset else args.plan
+        provenance = await self.provenance(plan)
+        names = list(plan.variables)
+        settings_names = list(plan.setting_variants)
+        axes = [plan.variables[k].values for k in names] + [plan.setting_variants[k] for k in settings_names]
+        count = math.prod(len(a) for a in axes)
+        if count > 10**12 or (args.mode == "all" and count > args.limit):
+            raise HTTPException(422, f"理论组合数 {count} 超出展开上限；缩小参数或采用随机采样")
+        indices = range(count) if args.mode == "all" else sorted(random.Random(args.seed).sample(range(count), min(count, args.limit)))
+        candidates, annotations, seen = [], {}, {}
+        for index in indices:
+            remaining, values = index, []
+            for axis in reversed(axes):
+                remaining, position = divmod(remaining, len(axis))
+                values.insert(0, axis[position])
+            bindings = dict(zip(names, values[:len(names)]))
+            def substitute(text):
+                return PLACEHOLDER.sub(lambda m: str(bindings[m.group(1)]), text)
+            selection, combo = substitute(plan.selection), substitute(plan.combo)
+            settings = {**plan.settings.model_dump(), **dict(zip(settings_names, values[len(names):]))}
+            variants = [(combo, combo == "1")] + ([("1", True)] if plan.include_baseline and combo != "1" else [])
+            for combo_value, baseline in variants:
+                candidate_id = f"super-{index + 1}{'-baseline' if baseline else ''}"
+                c = Candidate(client_item_id=candidate_id, alpha_type="SUPER", selection=selection,
+                              combo=combo_value, settings=settings)
+                h = fingerprint(c.platform_input())
+                annotations[candidate_id] = {"baseline": baseline, "parameters": bindings,
+                    "duplicate_of": seen.get(h), "request_hash": h}
+                seen.setdefault(h, candidate_id)
+                candidates.append(c.model_dump(mode="json"))
+        if len(candidates) > 10000:
+            raise HTTPException(422, "包含基线后超过 10000 项，请缩小候选数")
+        plan_reference = {"id": asset["id"], "version": asset["version"]} if asset else {}
+        source = Source(kind="superalpha", research_kind="superalpha", reference=plan.reference, hypothesis=plan.hypothesis,
+            superalpha_plan_id=args.plan_id, superalpha_plan_version=args.version,
+            selection_snapshot_ids=args.selection_snapshot_ids).model_dump(mode="json")
+        await validate_source(self.db, source, candidates)
+        experiment = ResearchExperiment(id=str(uuid4()), name=plan.name, kind="superalpha", hypothesis=plan.hypothesis,
+            inputs=[], parents=[], candidates=candidates, evidence={"plan": plan.model_dump(mode="json"),
+                "plan_reference": plan_reference, "provenance": provenance, "source": source,
+                "selection_snapshot_ids": args.selection_snapshot_ids, "annotations": annotations,
+                "combination_count": str(count), "mode": args.mode, "seed": args.seed})
+        self.db.add(experiment)
+        await self.db.flush()
+        result = await self.experiment(experiment.id)
+        return await self.remember("build_superalpha_candidates", args, digest, result, experiment.id)
+
+    async def experiment(self, experiment_id, limit=100, offset=0):
+        row = await self.db.get(ResearchExperiment, experiment_id)
+        if not row or row.kind != "superalpha":
+            raise HTTPException(404, "SUPER 研究记录不存在")
+        source = {**row.evidence["source"], "research_id": row.id}
+        visible = row.candidates[offset:offset + limit]
+        evidence = {**row.evidence, "annotations": {c["client_item_id"]: row.evidence["annotations"].get(c["client_item_id"], {}) for c in visible}}
+        return encode_snapshot({"id": row.id, "name": row.name, "kind": row.kind, "hypothesis": row.hypothesis,
+            "created_at": row.created_at, "evidence": evidence, "source": source,
+            "candidates": row.candidates[offset:offset + limit], "total": len(row.candidates),
+            "limit": limit, "offset": offset, "has_more": offset + limit < len(row.candidates)})
+
+    async def experiments(self, plan_id=None, limit=25, offset=0):
+        query = select(ResearchExperiment).where(ResearchExperiment.kind == "superalpha")
+        if plan_id:
+            query = query.where(ResearchExperiment.evidence["plan_reference"]["id"].as_string() == plan_id)
+        total = await self.db.scalar(select(func.count()).select_from(query.subquery()))
+        rows = await self.db.scalars(query.order_by(ResearchExperiment.created_at.desc(), ResearchExperiment.id).limit(limit).offset(offset))
+        return encode_snapshot({"items": [{"id": r.id, "name": r.name, "created_at": r.created_at,
+            "total": len(r.candidates)} for r in rows], "total": total, "limit": limit, "offset": offset})
+
+    async def preview(self, experiment_id, candidate_ids):
+        row = await self.db.get(ResearchExperiment, experiment_id)
+        await self.experiment(experiment_id)
+        selected = [c for c in row.candidates if c["client_item_id"] in set(candidate_ids)]
+        if len(selected) != len(set(candidate_ids)):
+            raise HTTPException(422, "候选不属于当前研究记录")
+        return await Backtests(self.db).preview(PreviewInput(inline=DraftInput(name=row.name,
+            candidates=selected, source={**row.evidence["source"], "research_id": row.id})), preserve_source=True)
+
+    async def selection_job(self, args):
+        if bool(args.plan_id) != bool(args.version):
+            raise HTTPException(422, "预览的方案来源需同时指定 ID 和版本")
+        if args.plan_id:
+            await self.assets.get(args.plan_id, args.version, "superalpha_plan")
+        await validate_settings(self.db, [args.settings])
+        account = await self.db.scalar(select(Account).where(Account.id == 1).with_for_update())
+        if not account or account.connection_status not in ("connected", "expired"):
+            raise HTTPException(409, "请先连接 WorldQuant")
+        payload = args.model_dump(mode="json")
+        jobs = await self.db.scalars(select(Job).where(Job.kind == "super_selection_preview",
+            Job.status.in_(("queued", "running", "waiting_auth", "waiting_connection"))))
+        job = next((j for j in jobs if j.payload == payload and not j.cancel_requested), None)
+        if not job:
+            job = Job(id=str(uuid4()), kind="super_selection_preview", payload=payload, total=1)
+            self.db.add(job)
+            await self.db.flush()
+        return {"job_id": job.id, "status": job.status, "read_with": "get_superalpha_selection"}
+
+    async def alpha(self, alpha_id, limit=25, offset=0):
+        from ..business import Business
+        from ..models import BacktestItem, BacktestResult
+        alpha = await self.db.get(Alpha, alpha_id)
+        if not alpha or alpha.alpha_type != "SUPER":
+            raise HTTPException(404, "SUPER Alpha 尚未导入")
+        item = await self.db.scalar(select(BacktestItem).join(BacktestResult, BacktestResult.item_id == BacktestItem.id)
+            .where(BacktestItem.alpha_id == alpha_id).order_by(BacktestResult.observed_at.desc()).limit(1))
+        components = await actual_components(self.db, item.id, limit, offset) if item else {
+            "status": "unknown", "complete": False, "source": "actual", "items": [], "total": 0}
+        if not item:
+            parsed = parse_components(alpha.raw.get("components", alpha.raw.get("selectedAlphas")))
+            components = {"source": "actual", "status": "available" if parsed["complete"] else "unknown",
+                "complete": parsed["complete"], "component_hash": parsed["component_hash"],
+                "reported_total": parsed["total"], "total": len(parsed["components"]), "warnings": parsed["warnings"],
+                "observed_at": alpha.synced_at, "items": parsed["components"][offset:offset + limit], "limit": limit, "offset": offset}
+        return {**await Business(self.db).get_alpha(alpha_id), "components": components,
+            "descriptions": {k: (alpha.raw.get(k) or {}).get("description", "")
+                if isinstance(alpha.raw.get(k), dict) else "" for k in ("selection", "combo")},
+            "sources": await Business(self.db).get_alpha_sources(alpha_id)}

backend/app/superalpha/settings.py

@@ -0,0 +1,45 @@
+"""Cached platform constraints shared by SUPER authoring and generic execution."""
+
+from fastapi import HTTPException
+
+from ..catalog.research_metadata import ResearchMetadata
+
+
+async def validate_settings(db, values):
+    """Reject known unsupported values; absent metadata is explicitly unknown, never approved."""
+    snapshot = await ResearchMetadata(db).get("settings")
+    rows = snapshot["content"].get("items", [])
+    if not snapshot["fetched_at"] or not rows:
+        return {"status": "unknown", "reason": "未缓存平台设置"}
+    incomplete = False
+    for settings in values:
+        value = settings.model_dump() if hasattr(settings, "model_dump") else settings
+        matches = [r for r in rows if all(r.get(k) == value.get(v) for k, v in (
+            ("instrument_type", "instrumentType"), ("region", "region"), ("universe", "universe"), ("delay", "delay")))]
+        if not matches:
+            raise HTTPException(422, "平台设置快照不支持当前 SUPER 地区 / Universe / Delay 组合")
+        failures = []
+        valid = False
+        for row in matches:
+            failed = []
+            if row.get("neutralizations") and value["neutralization"] not in row["neutralizations"]:
+                failed.append("neutralization")
+            for key, field in row.get("fields", {}).items():
+                if key not in value:
+                    continue
+                current = value[key]
+                if "choices" in field and current not in field["choices"]:
+                    failed.append(key)
+                if type(current) in (int, float) and (
+                    ("minimum" in field and current < field["minimum"]) or
+                    ("maximum" in field and current > field["maximum"])):
+                    failed.append(key)
+            if not failed:
+                valid = True
+                incomplete |= any(not row.get("fields", {}).get(key) for key in ("selectionLimit", "selectionHandling", "componentActivation"))
+                break
+            failures.extend(failed)
+        if not valid:
+            raise HTTPException(422, "平台设置快照不支持 SUPER 参数：" + "、".join(sorted(set(failures))))
+    return {"status": "partial" if incomplete else "cached", "fetched_at": snapshot["fetched_at"],
+            "reason": "部分 SUPER 设置范围未提供" if incomplete else "仅按缓存校验，仍需平台执行验证"}

backend/app/worldquant.py

@@ -468,6 +468,14 @@ class WqClient:
             "universe": scope.universe, "delay": scope.delay,
         })
 
+    async def run_super_selection(self, query):
+        """Read cnhk super-selection contract with bounded async retries and shared authentication."""
+        allowed = {"selection", "instrumentType", "region", "delay", "selectionLimit", "selectionHandling"}
+        if set(query) != allowed:
+            raise WqError("Selection 参数不完整或包含未知键", "invalid_selection")
+        return await self._read_json("GET", "/simulations/super-selection", params=query, allow_list=True,
+                                     wait_for_retry_header=True)
+
     async def research_setting_options(self):
         """Snapshot full setting choices for constrained research, including neutralization."""
         return await self._read_json("OPTIONS", "/simulations")

backend/migrations/versions/0021_superalpha.py

@@ -0,0 +1,39 @@
+"""SUPER candidates and immutable component evidence; retain all existing Alpha rows."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0021"
+down_revision = "0020"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column("backtest_items", sa.Column("alpha_type", sa.String(20), nullable=False, server_default="REGULAR"))
+    op.add_column("backtest_items", sa.Column("selection", sa.Text(), nullable=True))
+    op.add_column("backtest_items", sa.Column("combo", sa.Text(), nullable=True))
+    op.create_index("ix_backtest_items_alpha_type", "backtest_items", ["alpha_type"])
+    op.create_table("super_selection_snapshots",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("job_id", sa.String(36), sa.ForeignKey("sync_jobs.id"), unique=True),
+        sa.Column("item_id", sa.String(36), sa.ForeignKey("backtest_items.id"), unique=True),
+        sa.Column("source", sa.String(20), nullable=False),
+        sa.Column("request", sa.JSON(), nullable=False),
+        sa.Column("request_hash", sa.String(64), nullable=False),
+        sa.Column("component_hash", sa.String(64)),
+        sa.Column("components", sa.JSON(), nullable=False),
+        sa.Column("raw", sa.JSON(), nullable=False),
+        sa.Column("complete", sa.Boolean(), nullable=False),
+        sa.Column("total", sa.Integer()),
+        sa.Column("warnings", sa.JSON(), nullable=False),
+        sa.Column("observed_at", sa.DateTime(timezone=True), nullable=False))
+    for key in ("request_hash", "component_hash"):
+        op.create_index(f"ix_super_selection_snapshots_{key}", "super_selection_snapshots", [key])
+
+
+def downgrade():
+    op.drop_table("super_selection_snapshots")
+    op.drop_index("ix_backtest_items_alpha_type", table_name="backtest_items")
+    for key in ("combo", "selection", "alpha_type"):
+        op.drop_column("backtest_items", key)

backend/tests/backtest_fake.py

@@ -16,10 +16,16 @@ class Platform:
         self.detail_fail = False
         self.fail_child = None
         self.missing = False
+        self.selection_reads = []
+        self.selection_result = {"count": 2, "results": [{"id": "component1", "value": 0.3}, {"id": "component2", "value": 0.7}]}
+        self.actual_components = {"count": 2, "results": [{"id": "component1"}, {"id": "component2"}]}
         self.secret = "synthetic-platform-secret"
 
     def __call__(self, request):
         path = request.url.path
+        if path == "/simulations/super-selection":
+            self.selection_reads.append(dict(request.url.params))
+            return httpx.Response(200, json=self.selection_result)
         if path == "/authentication":
             return httpx.Response(201, json={})
         if path == "/simulations" and request.method == "POST":
@@ -39,29 +45,35 @@ class Platform:
             ids = []
             for i, item in enumerate(data):
                 child = parent if len(data) == 1 else f"{parent}c{i}"
-                aid = self.existing_alpha_ids[i] if self.existing_alpha_ids else f"alpha{parent}{i}"
+                aid = self.existing_alpha_ids[i] if self.existing_alpha_ids and item["type"] != "SUPER" else f"alpha{parent}{i}"
                 progress = {
                     "status": "COMPLETE",
                     "alpha": aid,
-                    "regular": item["regular"],
+                    "regular": item.get("regular", ""),
                     "settings": item["settings"],
                 }
                 if i == self.fail_child:
                     progress = {
                         "status": "FAILED",
-                        "regular": item["regular"],
+                        "regular": item.get("regular", ""),
                         "settings": item["settings"],
                         "message": "invalid expression",
                     }
                 self.simulations[child] = progress
                 self.alphas[aid] = {
                     "id": aid,
-                    "regular": {"code": item["regular"]},
+                    "regular": {"code": item.get("regular", "")},
                     "type": "REGULAR",
                     "settings": item["settings"],
                     "is": {"sharpe": None, "fitness": 0.8},
                     "status": "UNSUBMITTED",
                 }
+                if item["type"] == "SUPER":
+                    assert len(data) == 1, "SUPER must be submitted singly"
+                    self.simulations[child].update(type="SUPER", selection=item["selection"], combo=item["combo"])
+                    self.simulations[child].pop("regular", None)
+                    self.alphas[aid].update(type="SUPER", selection={"code": item["selection"], "description": "Selection rationale"}, combo={"code": item["combo"], "description": "Combo rationale"}, components=self.actual_components)
+                    self.alphas[aid].pop("regular", None)
                 ids.append(child)
             if len(data) > 1:
                 self.simulations[parent] = {

backend/tests/superalpha_postgres.py

@@ -0,0 +1,74 @@
+"""Disposable PostgreSQL compatibility/concurrency acceptance, no external platform calls.
+
+SUPER_TEST_DATABASE_URL must point to the local wq_superalpha_test database.
+"""
+import asyncio
+import os
+from urllib.parse import urlsplit
+
+import httpx
+from alembic import command
+from alembic.config import Config
+from cryptography.fernet import Fernet
+from sqlalchemy import select
+
+from app.alphas import upsert_alpha
+from app.config import Settings
+from app.main import create_app
+from app.models import BacktestItem, Research
+from app.superalpha.contracts import PlanSave
+from app.superalpha.service import SuperResearch
+from app.worldquant import WqClient
+from tests.backtest_fake import Platform
+from tests.test_backtests import execute, preview, setup, start
+from tests.test_superalpha import plan, test_plan_selection_build_versions_and_generic_run
+
+
+async def seed(settings):
+    app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(Platform())))
+    async with app.router.lifespan_context(app):
+        _, lane = await setup(app)
+        async with app.state.sessions.begin() as db:
+            await upsert_alpha(db, {"id": "legacy-super", "type": "SUPER", "selection": {"code": "turnover < 0.2"}, "combo": {"code": "1"}})
+            (await db.get(Research, "legacy-super")).note = "keep historical note"
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver", headers={"X-WQ-Request": "1"}) as client:
+            await client.post("/api/v1/auth/login", json={"username": "admin", "password": "synthetic-admin-only"})
+            run = await start(client, await preview(client), "legacy")
+            await execute(app, lane, run["backtest_run_id"])
+
+
+async def verify(settings):
+    app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(Platform())))
+    async with app.router.lifespan_context(app):
+        async with app.state.sessions() as db:
+            assert (await db.get(Research, "legacy-super")).note == "keep historical note"
+            item = await db.scalar(select(BacktestItem))
+            assert item.alpha_type == "REGULAR" and item.selection is None and item.combo is None
+            assert item.persistence_status == "saved"
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver", headers={"X-WQ-Request": "1"}) as client:
+            await client.post("/api/v1/auth/login", json={"username": "admin", "password": "synthetic-admin-only"})
+            await test_plan_selection_build_versions_and_generic_run(app, client)
+            imported = (await client.get("/api/v1/alphas?management_scope=super&q=legacy-super")).json()
+            assert imported["total"] == 1
+        async def save_same():
+            async with app.state.sessions.begin() as db:
+                return await SuperResearch(db).save(PlanSave(plan=plan(), idempotency_key="concurrent-save"))
+        first, second = await asyncio.gather(save_same(), save_same())
+        assert first == second
+    print("PostgreSQL: preserved REGULAR backtest and existing SUPER/notes; SUPER lifecycle and concurrent save replay passed")
+
+
+if __name__ == "__main__":
+    url = os.environ["SUPER_TEST_DATABASE_URL"]
+    parsed = urlsplit(url)
+    if parsed.hostname not in {"127.0.0.1", "localhost"} or parsed.path != "/wq_superalpha_test":
+        raise SystemExit("Refusing non-local/non-disposable database")
+    key = Fernet.generate_key().decode()
+    os.environ.update(DATABASE_URL=url, ADMIN_PASSWORD="synthetic-admin-only", ENCRYPTION_KEY=key, WQ_EMAIL="", WQ_PASSWORD="")
+    settings = Settings(_env_file=None, database_url=url, admin_password="synthetic-admin-only", encryption_key=key, enable_runner=False, public_origin="http://testserver")
+    config = Config("alembic.ini")
+    command.upgrade(config, "head")
+    asyncio.run(seed(settings))
+    command.downgrade(config, "0020")
+    command.upgrade(config, "head")
+    asyncio.run(verify(settings))

backend/tests/test_mcp.py

@@ -162,7 +162,7 @@ async def test_official_sdk_client_and_error_contract(mcp_app):
             async with ClientSession(streams[0], streams[1]) as client:
                 await client.initialize()
                 listed = await client.list_tools()
-                assert len(listed.tools) == 21
+                assert len(listed.tools) == 29
                 assert any(tool.name == "get_pyramid_distribution" for tool in listed.tools)
                 assert {"search_data_preparations", "get_data_preparation"} <= {t.name for t in listed.tools}
                 caps = await client.call_tool("get_research_capabilities", {})

backend/tests/test_superalpha.py

@@ -0,0 +1,210 @@
+"""SUPER public HTTP/MCP acceptance using isolated persistence and synthetic upstream."""
+import asyncio
+from copy import deepcopy
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy import func, select
+
+from app.backtests.contracts import Candidate
+from app.models import SimulationAttempt, SuperSelectionSnapshot
+from app.superalpha.evidence import parse_components
+from tests.test_backtests import PARAMS, candidate, execute, preview, setup, start
+from tests.test_mcp import credentials, invoke, mcp_app  # noqa: F401
+
+SETTINGS = PARAMS | {"selectionHandling": "POSITIVE", "selectionLimit": 100, "componentActivation": "IS"}
+
+
+def super_candidate(index=0, **changes):
+    return {"client_item_id": f"super-{index}", "alpha_type": "SUPER", "selection": f"turnover < {0.1 + index / 10}", "combo": "alpha", "settings": SETTINGS} | changes
+
+
+def plan(**changes):
+    return {"name": "Super 研究", "hypothesis": "降低组件换手", "selection": "turnover < {threshold}", "combo": "alpha", "variables": {"threshold": {"kind": "number", "values": [0.1, 0.2]}}, "settings": SETTINGS, "include_baseline": True} | changes
+
+
+async def test_plan_selection_build_versions_and_generic_run(app, logged_in):
+    platform, lane = await setup(app)
+    body = {"plan": plan(), "idempotency_key": "save1"}
+    saved = (await logged_in.post("/api/v1/superalpha/plans", json=body)).json()
+    assert saved["version"] == 1, saved
+    assert (await logged_in.post("/api/v1/superalpha/plans", json=body)).json() == saved
+    selection = {"selection": "turnover < 0.1", "settings": SETTINGS, "plan_id": saved["id"], "version": 1}
+    job = (await logged_in.post("/api/v1/superalpha/selections", json=selection)).json()
+    assert (await logged_in.post("/api/v1/superalpha/selections", json=selection)).json()["job_id"] == job["job_id"]
+    await app.state.runner.run_next()
+    snap = (await logged_in.get(f"/api/v1/superalpha/selections?job_id={job['job_id']}&limit=1")).json()
+    assert snap["complete"] and snap["has_more"] and snap["total"] == 2, snap
+    assert len(platform.selection_reads) == 1 and not platform.posts
+    assert set(platform.selection_reads[0]) == {"selection", "instrumentType", "region", "delay", "selectionLimit", "selectionHandling"}
+    build = {"plan_id": saved["id"], "version": 1, "selection_snapshot_ids": [snap["snapshot_id"]], "idempotency_key": "build1"}
+    exp = (await logged_in.post("/api/v1/superalpha/candidates", json=build)).json()
+    assert exp["total"] == 4, exp
+    assert (await logged_in.post("/api/v1/superalpha/candidates", json=build)).json() == exp
+    # A subset keeps the original experiment's evidence without claiming it applies to every candidate.
+    p = await logged_in.post(f"/api/v1/superalpha/experiments/{exp['id']}/preview", json={"candidate_ids": ["super-2", "super-2-baseline"]})
+    assert p.status_code == 200, p.text
+    assert p.json()["batch_count"] == 2 and not platform.posts
+    rid = (await start(logged_in, p.json()))["backtest_run_id"]
+    await execute(app, lane, rid)
+    result = (await logged_in.get(f"/api/v1/backtests/runs/{rid}/results")).json()
+    assert all(i["persistence_status"] == "saved" and i["alpha_type"] == "SUPER" for i in result["items"]), result
+    assert len(platform.posts) == 2 and all(len(p) == 1 for p in platform.posts)
+    item = result["items"][0]
+    actual = (await logged_in.get(f"/api/v1/backtests/items/{item['id']}/artifact?kind=components")).json()
+    assert actual["complete"] and actual["source"] == "actual" and actual["snapshot_id"] != snap["snapshot_id"]
+    assert actual["component_hash"] == snap["component_hash"]
+    alpha = (await logged_in.get(f"/api/v1/superalpha/alphas/{item['alpha_id']}")).json()
+    assert alpha["descriptions"]["combo"] == "Combo rationale"
+    assert alpha["sources"]["items"][0]["source"]["superalpha_plan_version"] == 1
+    changed = body | {"plan": plan(name="修订"), "plan_id": saved["id"], "version": 1, "idempotency_key": "save2"}
+    assert (await logged_in.post("/api/v1/superalpha/plans", json=changed)).json()["version"] == 2
+    assert (await logged_in.post("/api/v1/superalpha/plans", json=changed | {"idempotency_key": "save3"})).status_code == 409
+    assert (await logged_in.delete(f"/api/v1/superalpha/plans/{saved['id']}?version=2")).status_code == 200
+    assert (await logged_in.get(f"/api/v1/superalpha/plans/{saved['id']}?version=1")).json()["content"]["name"] == "Super 研究"
+    rows = (await logged_in.get(f"/api/v1/superalpha/experiments/{exp['id']}/results")).json()
+    assert rows["total"] == 2 and rows["items"][0]["pnl_fetched_at"] is None
+    async with app.state.sessions() as db:
+        assert await db.scalar(select(func.count()).select_from(SuperSelectionSnapshot)) == 3
+
+
+async def test_mixed_run_isolation_facets_export_and_strict_match(app, logged_in):
+    platform, lane = await setup(app)
+    p = await preview(logged_in, [candidate(0), candidate(1), super_candidate(), super_candidate(1)])
+    assert p["batch_count"] == 3
+    rid = (await start(logged_in, p))["backtest_run_id"]
+    await execute(app, lane, rid)
+    assert len(platform.posts) == 3
+    results = (await logged_in.get(f"/api/v1/backtests/runs/{rid}/results")).json()["items"]
+    assert all(i["persistence_status"] == "saved" for i in results), results
+    for scope, kind in (("super", "SUPER"), ("non_super", "REGULAR")):
+        page = (await logged_in.get(f"/api/v1/alphas?management_scope={scope}")).json()
+        assert page["total"] == 2 and all(i["alpha_type"] == kind for i in page["items"])
+        facets = (await logged_in.get(f"/api/v1/alphas/facets?management_scope={scope}")).json()
+        assert facets["alpha_type"] == [kind], facets
+        export = await logged_in.get(f"/api/v1/alphas/export?management_scope={scope}")
+        assert export.status_code == 200
+        assert all(i["alpha_id"] in export.text for i in results if i["alpha_type"] == kind)
+        assert all(i["alpha_id"] not in export.text for i in results if i["alpha_type"] != kind)
+    rid2 = (await start(logged_in, await preview(logged_in, [super_candidate(9)]), "mismatch"))["backtest_run_id"]
+    async with app.state.sessions() as db:
+        aid = await db.scalar(select(SimulationAttempt.id).where(SimulationAttempt.run_id == rid2))
+    await lane.step(aid)
+    next(reversed(platform.alphas.values()))["combo"]["code"] = "WRONG"
+    await lane.step(aid)
+    row = (await logged_in.get(f"/api/v1/backtests/runs/{rid2}/results")).json()["items"][0]
+    assert row["persistence_status"] != "saved"
+    assert (await logged_in.get("/api/v1/backtests/runs?alpha_type=SUPER")).json()["total"] == 2
+
+
+@pytest.mark.parametrize("reject", ["unknown", "missing_location", "rate"])
+async def test_super_reliability_no_unknown_resubmission(app, logged_in, reject):
+    platform, lane = await setup(app)
+    platform.reject = reject
+    rid = (await start(logged_in, await preview(logged_in, [super_candidate()])))["backtest_run_id"]
+    async with app.state.sessions() as db:
+        aid = await db.scalar(select(SimulationAttempt.id).where(SimulationAttempt.run_id == rid))
+    for _ in range(app.state.settings.retry_attempts + 1):
+        await lane.step(aid)
+        await asyncio.sleep(0.02)
+    if reject == "rate":
+        assert len(platform.posts) == app.state.settings.retry_attempts
+    else:
+        await lane.start()
+        await lane.stop()
+        await logged_in.post(f"/api/v1/backtests/runs/{rid}/control", json={"action": "recover", "version": 1})
+        await lane.step(aid)
+        assert len(platform.posts) == 1
+
+
+async def test_unknown_components_and_recovery(app, logged_in):
+    platform, lane = await setup(app)
+    platform.actual_components = None
+    platform.detail_fail = True
+    rid = (await start(logged_in, await preview(logged_in, [super_candidate()])))["backtest_run_id"]
+    ids = await execute(app, lane, rid)
+    platform.detail_fail = False
+    await logged_in.post(f"/api/v1/backtests/runs/{rid}/control", json={"action": "recover", "version": 1})
+    await lane.step(ids[0])
+    item = (await logged_in.get(f"/api/v1/backtests/runs/{rid}/results")).json()["items"][0]
+    assert item["persistence_status"] == "saved" and len(platform.posts) == 1
+    components = (await logged_in.get(f"/api/v1/backtests/items/{item['id']}/artifact?kind=components")).json()
+    assert components["complete"] is False and components["component_hash"] is None
+
+
+async def test_sampling_validation_and_component_evidence(app, logged_in):
+    await setup(app)
+    for invalid in (super_candidate(selection=" "), super_candidate(combo=""), super_candidate(expression="close"), super_candidate(settings=PARAMS)):
+        with pytest.raises(ValueError):
+            Candidate.model_validate(invalid)
+    assert (await logged_in.get("/api/v1/superalpha/selections")).status_code == 422
+    assert parse_components({"count": 0, "results": []})["complete"]
+    for raw in (["a"], {"count": 2, "results": ["a"]}, {"count": 2, "results": ["a", "a"]}, {"count": 1, "results": ["a"], "next": "next"}):
+        assert not parse_components(raw)["complete"]
+    complete = parse_components({"count": 1, "results": ["a"], "warnings": ["synthetic warning"]})
+    assert complete["complete"] and complete["warnings"] == ["synthetic warning"]
+    args = {"plan": plan(variables={"threshold": {"kind": "number", "values": list(range(100))}}), "mode": "random", "limit": 5, "seed": 42, "idempotency_key": "random1"}
+    a = (await logged_in.post("/api/v1/superalpha/candidates", json=args)).json()
+    b = (await logged_in.post("/api/v1/superalpha/candidates", json=args | {"idempotency_key": "random2"})).json()
+    assert a["candidates"] == b["candidates"] and a["total"] == 10
+    bad = deepcopy(args)
+    bad["plan"]["setting_variants"] = {"selectionLimit": [0]}
+    assert (await logged_in.post("/api/v1/superalpha/candidates", json=bad)).status_code == 422
+
+
+async def test_mcp_interop_permissions_inline_and_sources(mcp_app):  # noqa: F811
+    app = mcp_app
+    principal, _ = await credentials(app, {"research:read", "research:write", "research:refresh", "backtests:execute"})
+    saved = await invoke(app, principal, "save_superalpha_plan", {"plan": plan(), "idempotency_key": "mcp-plan"})
+    built = await invoke(app, principal, "build_superalpha_candidates", {"plan_id": saved["id"], "version": 1, "idempotency_key": "mcp-build"})
+    assert not built["starts_backtests"] and built["total"] == 4
+    assert (await invoke(app, principal, "get_superalpha_plan", {"experiment_id": built["id"], "limit": 1}))["has_more"]
+    submit = {"name": "MCP SUPER", "candidates": built["candidates"][:1], "source": {"research_id": built["id"], "superalpha_plan_id": saved["id"], "superalpha_plan_version": 1}, "duplicate_policy": "rerun", "idempotency_key": "mcp-run"}
+    result = await invoke(app, principal, "submit_backtests", submit)
+    assert result == await invoke(app, principal, "submit_backtests", submit)
+    rid = result["backtest_run_id"]
+    await execute(app, app.state.runner.backtests, rid)
+    results = await invoke(app, principal, "get_backtest_results", {"run_id": rid})
+    item = results["items"][0]
+    assert item["persistence_status"] == "saved", results
+    alpha = await invoke(app, principal, "get_superalpha", {"alpha_id": item["alpha_id"]})
+    assert alpha["alpha_type"] == "SUPER" and "#superalphas?" in alpha["web_url"]
+    forged = deepcopy(submit)
+    forged["idempotency_key"] = "forged"
+    forged["candidates"][0]["combo"] = "WRONG"
+    assert (await app.state.mcp.invoke(principal, "submit_backtests", forged)).is_error
+    direct = submit | {"source": {}, "candidates": [super_candidate(9)], "idempotency_key": "direct"}
+    assert (await invoke(app, principal, "submit_backtests", direct))["source"]["superalpha_plan_id"] is None
+    readonly, _ = await credentials(app, {"research:read"})
+    with pytest.raises(HTTPException) as exc:
+        await app.state.mcp.invoke(readonly, "save_superalpha_plan", {"plan": plan(), "idempotency_key": "denied"})
+    assert exc.value.status_code == 403
+
+
+async def test_metadata_stage_constraints_selection_recovery_and_cancel(app, logged_in):
+    from app.catalog.research_metadata import ResearchMetadata
+    from app.models import Job
+    from app.superalpha.jobs import run_selection
+    platform, _ = await setup(app)
+    async with app.state.sessions.begin() as db:
+        await ResearchMetadata(db).publish("settings", "settings", {"items": [{"instrument_type": "EQUITY", "region": "USA", "delay": 1, "universe": "TOP3000", "neutralizations": ["INDUSTRY"], "fields": {"selectionLimit": {"maximum": 50}}}]})
+        await ResearchMetadata(db).publish("operators", "operators", {"items": [
+            {"name": "combo_a", "category": "Combo", "scope": ["COMBO"]},
+            {"name": "not_combo", "category": "Other", "scope": ["UNKNOWN_COMBO"]},
+            {"name": "unspecified", "category": "Other", "scope": None}]})
+    assert (await logged_in.post("/api/v1/superalpha/plans", json={"plan": plan(), "idempotency_key": "bad-settings"})).status_code == 422
+    p = await logged_in.post("/api/v1/backtests/previews", json={"inline": {"name": "bad", "candidates": [super_candidate()]}})
+    assert p.status_code == 422
+    assert [o["name"] for o in (await logged_in.get("/api/v1/catalog/operators?stage=COMBO")).json()["items"]] == ["combo_a"]
+    payload = {"selection": "turnover", "settings": SETTINGS | {"selectionLimit": 50}}
+    job = (await logged_in.post("/api/v1/superalpha/selections", json=payload)).json()
+    await app.state.runner.run_next()
+    await run_selection(app.state.runner, job["job_id"], payload)  # Restart after commit preserves evidence.
+    assert len(platform.selection_reads) == 1
+    cancelled = (await logged_in.post("/api/v1/superalpha/selections", json=payload)).json()
+    async with app.state.sessions.begin() as db:
+        (await db.get(Job, cancelled["job_id"])).cancel_requested = True
+    await app.state.runner.run_next()
+    async with app.state.sessions() as db:
+        assert (await db.get(Job, cancelled["job_id"])).status == "cancelled"
+        assert await db.scalar(select(func.count()).select_from(SuperSelectionSnapshot)) == 1