From a6e36e50eccd8023042dd9f8544e45a3b1720499 Mon Sep 17 00:00:00 2001 From: yuxuanhui Date: Tue, 8 Sep 2026 22:42:29 +0800 Subject: [PATCH] feat: compose native research workflows in QuantFlow --- .../research-migration/acceptance/stage-4.md | 46 ++ .../issues/01-implementation.md | 6 +- backend/app/ai/contracts.py | 1 + backend/app/research/experiments.py | 9 +- backend/app/research/routes.py | 54 +- backend/app/research/runtime.py | 270 ++++-- backend/app/research/workflows.py | 75 +- backend/app/research/workspace_tools.py | 18 + backend/tests/research_quantflow_postgres.py | 121 +++ backend/tests/test_quantflow.py | 286 +++++++ frontend/src/App.tsx | 25 +- frontend/src/ai/types.ts | 3 +- frontend/src/ai/workspace.ts | 2 + frontend/src/components/AppSidebar.tsx | 1 + frontend/src/research/FlowCanvas.tsx | 246 ++++++ frontend/src/research/FlowLaunchForm.tsx | 33 +- frontend/src/research/PipelinePage.tsx | 2 +- frontend/src/research/QuantFlowPage.tsx | 775 ++++++++++++++++++ frontend/src/research/ResearchToolCard.tsx | 3 +- frontend/src/research/flowTypes.ts | 1 + frontend/src/research/workspace.css | 146 ++++ frontend/tests/quantflow.spec.ts | 236 ++++++ 22 files changed, 2274 insertions(+), 85 deletions(-) create mode 100644 .scratch/research-migration/acceptance/stage-4.md create mode 100644 backend/tests/research_quantflow_postgres.py create mode 100644 backend/tests/test_quantflow.py create mode 100644 frontend/src/research/FlowCanvas.tsx create mode 100644 frontend/src/research/QuantFlowPage.tsx create mode 100644 frontend/tests/quantflow.spec.ts diff --git a/.scratch/research-migration/acceptance/stage-4.md b/.scratch/research-migration/acceptance/stage-4.md new file mode 100644 index 0000000..4fe10b7 --- /dev/null +++ b/.scratch/research-migration/acceptance/stage-4.md @@ -0,0 +1,46 @@ +# 第四阶段验收：QuantFlow 原生研究画布 + +日期：2026-09-08。范围：流程画布、固定流程版本、原生节点及共用研究执行。所有平台、账户和模型响应均为隔离合成数据，没有调用真实模型或 WorldQuant。 + +## 交付 + +- 研究编排分组增加 QuantFlow，保留原侧栏、最近访问和 AI 助手。流程库、画布、节点配置、连线、版本与运行记录可直接使用。 +- 画布支持拖动及方向键移动节点、缩放、适应宽度，配置素材版本、研究说明、变体方法、筛选结论和迭代轮数。草稿保留本地基准版本，冲突更新返回 409；历史版本可查看或另存。 +- 原生节点包括固定输入、特征方案、模板生成/增强、模板展开、两类变体、回测、规则评估、筛选、条件、汇总、有界迭代。固定六阶段研究可复制为内置配方。 +- 保存及启动都验证节点唯一性、连接端点、输入输出类型、普通连接无环、可达性、配置和迭代约束。首版一个流程最多 50 节点、100 连线；一个终点迭代节点重复整个流程，最多 100 轮，不能超过本次授权轮数。 +- 运行固定工作流版本、节点引用的模板/特征版本、输入、种子、规则和预算。启动不含模型节点的流程无需配置模型；所有预算仍须为有限正整数。 +- 模板、特征及结构变体复用统一表达式模块；设置变体冻结种子表达式及原设置，按独立目标输入计算合法设置，回测前再次核对授权范围。候选上限与采样种子保留。 +- 条件按规则报告的 pass/review/block 选择出边，未命中分支和空筛选不会执行后续模拟。汇总保存产物引用，避免递归复制整个上游图。 +- 完全复用研究运行、步骤、预算、预览、回测调度和评估实现；重复事件不增加模拟，刷新后恢复运行和节点产物显示。 +- 模型返回后先保存产物；特征转换和结构变体展开在下一次活跃步骤进行。暂停、停止、账户改变期间只收集已发结果，不继续转换或展开；恢复不重复已完成模型调用。 +- AI 可读取已有流程版本、研究运行及预算；普通 Chatbox 不能借来源字段获得自动执行授权。 + +## 验收结果 + +| 验收项 | 结果 | +|---|---| +| 非法节点图、连线、循环和配置 | 通过；重复边、断连、类型不符、普通循环、布尔迭代上限、未知脚本配置、非法筛选和素材版本返回 422 | +| 保存后修改模板和流程 | 通过；旧运行继续使用 v1，模板更新为无效表达式不影响已启动版本 | +| 三路条件与汇总 | 通过；仅一个匹配分支执行，其余两个跳过；汇总保存多个上游引用 | +| 固定特征 → 模板 → 展开 → 回测 → 筛选 → 迭代 | 通过；两轮共 4 条模拟、0 次模型调用，特征原版本可追溯 | +| 模型特征与结构变体 | 通过；各使用一次模型调用，保存特征/模板、候选及种子关系 | +| 市场与设置变体 | 通过；启动后种子被改动仍使用原表达式，目标输入和股票池严格对应，没有调用模型 | +| 空筛选、重复完成事件 | 通过；不发后续模拟，不重复回测 | +| 模型期间暂停、停止、账户变化 | 通过；模型产物持久化但没有新候选/回测，暂停后恢复复用产物，模型调用数保持 1 | +| 后端全量回归 | 最终 220 passed，日志 `/tmp/wq-stage4-full-backend-final.log` | +| 静态检查和前端构建 | Ruff、TypeScript、构建及 diff whitespace 检查通过；保留依赖 lottie-web 原有 eval 构建警告 | +| 浏览器画布闭环 | 通过；新建 8 节点、连线、键盘移动、保存 v1/v2、预算确认、固定 v2 启动、分支和刷新恢复，模拟 2 条、模型调用 0 次 | +| 浏览器回归 | 8 passed，覆盖 QuantFlow、固定流水线、特征/成果、侧栏、4 项 AI 行为；日志 `/tmp/wq-stage4-browser-regression.log` | +| PostgreSQL 17 与备份恢复 | 通过；从第三阶段备份恢复，Alembic head 0008 无新增迁移差异，两轮原生图执行与重放正常；pg_dump/pg_restore 后完整运行/步骤快照摘要及旧研究备注一致。脚本 `backend/tests/research_quantflow_postgres.py`，日志 `/tmp/wq-stage4-postgres.log` | + +画布连线路由和适应宽度调整后，单独复验 1 passed，日志 `/tmp/wq-stage4-canvas-final.log`。 + +最终截图：`/tmp/wq-stage-four-canvas.png`、`/tmp/wq-stage-four-run.png`。主代理已检查画布、运行状态、来源入口和中文界面。 + +独立只读核验已完成。其指出暂停后的模型回调仍执行变体展开，已拆成持久的模型产物与后续转换步骤，增加暂停、停止及账户变化回归。全量测试过程中发现测试辅助函数漏导入和 AI 工具重复注册，均已修正，以上列出的是最终通过结果。 + +## 边界与待联调 + +首版迭代重复整个流程，普通节点各接受一个上游，汇总允许多个上游；不提供任意脚本、CLI、外部事件自动启动或正式 Alpha 提交。流程编辑与启动分开，扩大预算和范围须重新确认新运行。 + +真实 WorldQuant 认证、元数据协议及模拟联调仍待当前对话确认授权。自动审批此前拒绝了读取 account.json 后向官方平台联调的提权命令，理由是未将任务目标记录中的授权认可为当前用户消息授权；没有绕过审批。本轮合成测试不能证明真实字段可用性和平台合法设置协议已联调通过。 diff --git a/.scratch/research-migration/issues/01-implementation.md b/.scratch/research-migration/issues/01-implementation.md index dbb157b..57b2484 100644 --- a/.scratch/research-migration/issues/01-implementation.md +++ b/.scratch/research-migration/issues/01-implementation.md @@ -7,8 +7,8 @@ Status: ready-for-agent - [x] 一：算子、模板、表达式模块、两类变体与回测来源闭环。 - [x] 二：特征方案、保存视图、关系、比较和评估。 - [x] 三：固定自动研究、预算、持久步骤、恢复。 -- [ ] 四：原生 QuantFlow 画布与共用执行。 -- [ ] 验证：后端、前端、浏览器、PostgreSQL 迁移与恢复。 +- [x] 四：原生 QuantFlow 画布与共用执行。 +- [x] 验证：后端、前端、浏览器、PostgreSQL 迁移与恢复。 ## Comments @@ -19,3 +19,5 @@ Status: ready-for-agent 第二阶段已通过后端、前端、浏览器和 PostgreSQL 迁移/恢复验证；报告见 `../acceptance/stage-2.md`。独立 Alpha 列表布局改动继续留在工作区，本次仅提交保存视图接入。 第三阶段固定研究及有限授权已验收，报告见 `../acceptance/stage-3.md`。 + +第四阶段原生 QuantFlow 已完成本地后端、前端、浏览器及 PostgreSQL 验收，报告见 `../acceptance/stage-4.md`。四阶段实现与模拟验证完成；真实 WorldQuant 联调未执行，限制详见各阶段报告。 diff --git a/backend/app/ai/contracts.py b/backend/app/ai/contracts.py index 1a01090..5955456 100644 --- a/backend/app/ai/contracts.py +++ b/backend/app/ai/contracts.py @@ -46,6 +46,7 @@ class PageContext(Contract): "features", "variants", "pipeline", + "quantflow", ] = "alphas" research_run_id: str | None = Field(default=None, max_length=36) research_asset_id: str | None = Field(default=None, max_length=36) diff --git a/backend/app/research/experiments.py b/backend/app/research/experiments.py index d8c96b3..e57c69a 100644 --- a/backend/app/research/experiments.py +++ b/backend/app/research/experiments.py @@ -310,8 +310,10 @@ class Experiments: preserve_source=True, ) - async def setting_variants(self, body): - parents = await self.parents([body.alpha_id], []) + async def setting_variants(self, body, *, parent_snapshot=None, extra_evidence=None, kind="variant"): + parents = ( + [parent_snapshot] if parent_snapshot is not None else await self.parents([body.alpha_id], []) + ) original = parents[0] base = SimulationSettings.model_validate(original["settings"]) expression = original["expression"] @@ -362,12 +364,13 @@ class Experiments: ) return await self.save( f"{body.alpha_id} · 设置变体", - "variant", + kind, body.hypothesis, snapshots, parents, candidates, { + **(extra_evidence or {}), "method": "settings", "rejected": rejected, "operators_snapshot": operators_snapshot, diff --git a/backend/app/research/routes.py b/backend/app/research/routes.py index 9741b98..3ac0ba5 100644 --- a/backend/app/research/routes.py +++ b/backend/app/research/routes.py @@ -22,6 +22,7 @@ from .workspace_contracts import ( ImportCommit, ImportPreview, SettingVariants, + WorkflowSpec, ) router = APIRouter(prefix="/api/v1/research", tags=["research"], dependencies=[Depends(require_auth)]) @@ -41,7 +42,7 @@ async def assets( limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0), ): - if kind not in ("template", "feature", "view"): + if kind not in ("template", "feature", "view", "workflow"): raise HTTPException(422, "当前素材类型尚未开放") async with request.app.state.sessions() as db: return await Assets(db).list(kind, q, limit, offset) @@ -49,7 +50,7 @@ async def assets( @router.post("/assets", status_code=201) async def save_asset(body: AssetWrite, request: Request): - if body.kind not in ("template", "feature", "view"): + if body.kind not in ("template", "feature", "view", "workflow"): raise HTTPException(422, "当前素材类型尚未开放") async with request.app.state.sessions.begin() as db: return await Assets(db).save(body) @@ -63,7 +64,7 @@ async def asset(asset_id: str, request: Request, version: int | None = Query(Non @router.put("/assets/{asset_id}") async def update_asset(asset_id: str, body: AssetWrite, request: Request): - if body.kind not in ("template", "feature", "view"): + if body.kind not in ("template", "feature", "view", "workflow"): raise HTTPException(422, "当前素材类型尚未开放") async with request.app.state.sessions.begin() as db: return await Assets(db).save(body, asset_id) @@ -240,21 +241,42 @@ async def fixed_recipe(): async def start_flow(body: FlowStart, request: Request): from .workflows import Workflows - if body.workflow_id: - raise HTTPException(422, "自定义流程将在 QuantFlow 阶段开放") async with request.app.state.sessions.begin() as db: - config = await request.app.state.ai.config(db) - result = await Workflows(db).start(body, config.revision) + from .workflows import fixed_workflow + from .workspace_contracts import WorkflowSpec + + graph = ( + WorkflowSpec.model_validate( + (await Assets(db).get(body.workflow_id, body.workflow_version, "workflow"))["content"] + ) + if body.workflow_id + else fixed_workflow(body.budget.max_rounds) + ) + needs_model = any( + n.type == "generate" + or (n.type == "feature" and not n.config.get("asset_id")) + or (n.type == "variant" and n.config.get("method", "structure") == "structure") + for n in graph.nodes + ) + config = await request.app.state.ai.config(db) if needs_model else None + result = await Workflows(db).start(body, config.revision if config else None) request.app.state.research.wake.set() return result @router.get("/flows/runs") -async def flow_runs(request: Request, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)): +async def flow_runs( + request: Request, + limit: int = Query(25, ge=1, le=100), + offset: int = Query(0, ge=0), + kind: str | None = None, +): from .workflows import Workflows async with request.app.state.sessions() as db: - return await Workflows(db).list(limit, offset) + if kind not in (None, "pipeline", "quantflow"): + raise HTTPException(422, "未知研究运行类型") + return await Workflows(db).list(limit, offset, kind) @router.get("/flows/runs/{run_id}") @@ -276,3 +298,17 @@ async def control_flow(run_id: str, body: FlowControl, request: Request): request.app.state.research.wake.set() request.app.state.runner.backtests.wake.set() return result + + +@router.get("/flows/nodes") +async def flow_nodes(): + from .workflows import NODE_TYPES + + return {"items": [{"type": key, **value} for key, value in NODE_TYPES.items()]} + + +@router.post("/flows/validate") +async def validate_flow(body: WorkflowSpec): + from .workflows import validate_graph + + return {"valid": True, "order": validate_graph(body)} diff --git a/backend/app/research/runtime.py b/backend/app/research/runtime.py index 4d315ac..3664e54 100644 --- a/backend/app/research/runtime.py +++ b/backend/app/research/runtime.py @@ -6,21 +6,32 @@ Backtest source fields are provenance, never a grant to execute automatically. import asyncio import logging +import random from collections import defaultdict from fastapi import HTTPException from sqlalchemy import select from sqlalchemy.exc import SQLAlchemyError -from ..backtests.contracts import ControlInput, StartInput +from ..backtests.contracts import ControlInput, SimulationSettings, StartInput from ..backtests.service import Backtests, uid from ..models import Account, ResearchFlowRun, ResearchStepRun, now from .assets import Assets from .evaluations import Evaluations -from .experiments import Experiments +from .experiments import Experiments, scope_of +from .features import Features from .model import request_model from .workflows import validate_graph -from .workspace_contracts import AssetWrite, EvaluateInput, Expansion, Generation, TemplateSpec, WorkflowSpec +from .workspace_contracts import ( + AssetWrite, + EvaluateInput, + Expansion, + FeatureSpec, + Generation, + SettingVariants, + TemplateSpec, + WorkflowSpec, +) logger = logging.getLogger(__name__) ACTIVE = ("queued", "running") @@ -162,7 +173,8 @@ class ResearchRuntime: model_work = await self.prepare(run_id) if model_work: step_id, context, revision = model_work - result, evidence = await request_model(self.ai, context, TemplateSpec, revision) + output_type = FeatureSpec if context.get("method") == "feature" else TemplateSpec + result, evidence = await request_model(self.ai, context, output_type, revision) await self.finish_model(run_id, step_id, result, evidence) except HTTPException as exc: await self.fail(run_id, str(exc.detail)) @@ -271,7 +283,7 @@ class ResearchRuntime: else None ) step.output = { - **(previous.output if previous else {}), + **(previous.output if previous else {"template": run.authorization.get("template")}), "type": "context", "input_ids": run.authorization["input_ids"], } @@ -287,50 +299,45 @@ class ResearchRuntime: } else: return await self.reserve_model(db, run, step, data, node) + elif node.type == "feature": + reference = run.authorization.get("node_assets", {}).get(key) + if step.output.get("feature"): + ref = step.output["feature"] + reference = await Assets(db).get(ref["id"], ref["version"], "feature") + if reference: + template = await Features(db).to_template(reference["id"], reference["version"]) + step.output = { + **data, + **step.output, + "type": "context", + "feature": {"id": reference["id"], "version": reference["version"]}, + "input_ids": reference["content"]["input_ids"], + "template": {"id": template["id"], "version": template["version"]}, + } + else: + return await self.reserve_model(db, run, step, data, node) elif node.type == "expand": - if not data.get("template"): - raise HTTPException(422, "展开节点没有固定模板版本") - body = Expansion( - asset_id=data["template"]["id"], - version=data["template"]["version"], - input_ids=run.authorization["input_ids"], - hypothesis=run.authorization["hypothesis"], - settings=run.authorization["settings"], - mode="random", - limit=run.authorization["batch_candidates"], - seed=run.authorization["seed"] + run.round, - parent_alpha_ids=run.authorization["parent_alpha_ids"], - parent_experiment_ids=data.get("parent_experiment_ids", []), + reference = run.authorization.get("node_assets", {}).get(key) or data.get("template") + await self.expand(db, run, step, {**data, "template": reference}) + return + elif node.type == "variant": + if node.config.get("method", "structure") == "structure": + if step.output.get("template"): + await self.expand(db, run, step, step.output) + return + return await self.reserve_model(db, run, step, data, node) + experiment = await Experiments(db).setting_variants( + SettingVariants( + alpha_id=run.authorization["parent_alpha_ids"][0], + input_ids=run.authorization["input_ids"], + hypothesis=run.authorization["hypothesis"], + ), + parent_snapshot=run.authorization["parents"][0], + extra_evidence={"flow_run_id": run.id, "node_id": key, "round": run.round}, + kind=run.authorization["kind"], ) - frozen_parents = run.authorization["parents"] + await Experiments(db).parents( - [], body.parent_experiment_ids - ) - experiment = await Experiments(db).create( - body, - "pipeline", - { - "flow_run_id": run.id, - "node_id": key, - "round": run.round, - "method": "pipeline", - "authorized_seed_snapshots": run.authorization["parents"], - }, - parent_snapshots=frozen_parents, - ) - step.output = { - "type": "candidates", - "experiment_id": experiment["id"], - "candidate_ids": [ - c["client_item_id"] - for c in experiment["candidates"] - if c["validation"]["status"] == "valid" - ], - "template": data["template"], - } - if not step.output["candidate_ids"]: - step.status, step.error = "blocked", "候选均未通过本地校验，请核实字段、算子和设置" - halt(run, "needs_review", step.error) - return + self.candidates(run, step, experiment, {}) + return elif node.type == "backtest": await self.backtest(db, run, step, data) return @@ -349,6 +356,55 @@ class ResearchRuntime: "experiment_id": data["experiment_id"], "backtest_run_id": data["backtest_run_id"], } + elif node.type == "condition": + step.output = {**data, "type": "evaluation"} + elif node.type == "filter": + report = (await Evaluations(db).get(data["evaluation_id"]))["report"] + ids = [ + r["client_item_id"] + for r in report["records"] + if r["verdict"] in node.config.get("verdicts", ["pass"]) + ] + experiment = await Experiments(db).get(data["experiment_id"]) + template = experiment["evidence"].get("template") + step.output = { + **data, + "type": "candidates", + "candidate_ids": ids, + "template": {"id": template["id"], "version": template["version"]} + if template + else None, + } + if not ids: + step.status, step.updated_at = "skipped", now() + return + elif node.type == "summarize": + # References keep joins bounded, without recursively copying the upstream graph. + step.output = { + "type": "summary", + "artifacts": [ + { + "step_id": steps[e.source].id, + "node_id": e.source, + **{ + k: steps[e.source].output[k] + for k in ( + "type", + "template", + "feature", + "experiment_id", + "evaluation_id", + "backtest_run_id", + "verdict", + ) + if k in steps[e.source].output + }, + } + for e in upstream_edges + if steps[e.source].status != "skipped" + and (not e.branch or steps[e.source].output.get("verdict") == e.branch) + ], + } elif node.type == "iterate": step.output = { **data, @@ -370,10 +426,16 @@ class ResearchRuntime: generation = Generation( name=run.name, hypothesis=run.authorization["hypothesis"], - input_ids=run.authorization["input_ids"], + input_ids=data.get("input_ids", run.authorization["input_ids"]), + method="feature" + if node.type == "feature" + else "structure" + if node.type == "variant" + else "template", parent_experiment_ids=[data["experiment_id"]] if data.get("experiment_id") else [], ) context = await Experiments(db).generation_context(generation) + context["method"] = generation.method context["parents"] = run.authorization["parents"] + context["parents"] context["operators"] = run.authorization["operators_snapshot"]["content"]["items"][:100] context["node_prompt"] = node.config.get("prompt", "") @@ -414,9 +476,16 @@ class ResearchRuntime: step = await db.get(ResearchStepRun, step_id) if not step or step.status != "running": return + if isinstance(result, FeatureSpec) and set(result.input_ids) != set( + [i["id"] for i in step.output["context"]["inputs"]] + ): + raise HTTPException(422, "模型不能改变已固定的输入范围") # A paused/stopped run may collect this already-issued model output, but cannot advance. asset = await Assets(db).save( - AssetWrite(kind="template", content=result.model_dump(mode="json")), + AssetWrite( + kind="feature" if isinstance(result, FeatureSpec) else "template", + content=result.model_dump(mode="json"), + ), provenance={ "flow_run_id": run_id, "step_id": step_id, @@ -424,18 +493,92 @@ class ResearchRuntime: "context": step.output["context"], }, ) + feature = ( + {"id": asset["id"], "version": asset["version"]} if isinstance(result, FeatureSpec) else None + ) mark_model_attempt(step, "completed") step.output = { "model_attempts": step.output.get("model_attempts", []), - "type": "template", - "template": {"id": asset["id"], "version": asset["version"]}, + "type": "context" if feature else "template", + "feature": feature, + "input_ids": result.input_ids + if feature + else [i["id"] for i in step.output["context"]["inputs"]], + "template": None if feature else {"id": asset["id"], "version": asset["version"]}, "parent_experiment_ids": step.output.get("parent_experiment_ids", []), "generation": evidence, "reserved_call": step.output["reserved_call"], } - step.status, step.updated_at = "completed", now() + node = next(n for n in run.definition["nodes"] if n["id"] == step.node_id) + # Keep post-processing durable and separate: pause/account changes are checked again + # before converting features or expanding variants on the next active tick. + step.status = "generated" if node["type"] in ("variant", "feature") else "completed" + step.updated_at = now() changed(run) + def candidates(self, run, step, experiment, data): + step.output = { + **data, + "type": "candidates", + "experiment_id": experiment["id"], + "candidate_ids": [ + c["client_item_id"] for c in experiment["candidates"] if c["validation"]["status"] == "valid" + ], + } + ids = step.output["candidate_ids"] + maximum = run.authorization["batch_candidates"] + if len(ids) > maximum: + step.output = { + **step.output, + "candidate_ids": random.Random(run.authorization["seed"] + run.round).sample(ids, maximum), + } + step.status, step.updated_at = "completed", now() + if not step.output["candidate_ids"]: + step.status, step.error = "blocked", "候选均未通过本地校验，请核实字段、算子和设置" + if run.status in ACTIVE: + halt(run, "needs_review", step.error) + + async def expand(self, db, run, step, data): + reference = data.get("template") + if not reference: + raise HTTPException(422, "展开节点没有固定模板版本") + scope = scope_of(SimulationSettings.model_validate(run.authorization["settings"])) + ids = [ + i["id"] + for i in run.authorization["inputs"] + if i["scope"] == scope and i["id"] in data.get("input_ids", run.authorization["input_ids"]) + ] + if not ids: + raise HTTPException(422, "模板展开需要与基础设置匹配的固定输入") + body = Expansion( + asset_id=reference["id"], + version=reference["version"], + input_ids=ids, + hypothesis=run.authorization["hypothesis"], + settings=run.authorization["settings"], + mode="random", + limit=run.authorization["batch_candidates"], + seed=run.authorization["seed"] + run.round, + parent_alpha_ids=run.authorization["parent_alpha_ids"], + parent_experiment_ids=data.get("parent_experiment_ids", []), + ) + frozen_parents = run.authorization["parents"] + await Experiments(db).parents( + [], body.parent_experiment_ids + ) + experiment = await Experiments(db).create( + body, + run.authorization["kind"], + { + "flow_run_id": run.id, + "node_id": step.node_id, + "round": run.round, + "method": run.authorization["kind"], + "authorized_seed_snapshots": run.authorization["parents"], + }, + parent_snapshots=frozen_parents, + ) + self.candidates(run, step, experiment, data) + async def backtest(self, db, run, step, data): service = Backtests(db) if step.backtest_run_id: @@ -470,12 +613,27 @@ class ResearchRuntime: ): raise HTTPException(409, "研究候选预览不再匹配保存的授权步骤") experiment = await Experiments(db).get(step.output["experiment_id"]) - if experiment["evidence"].get("flow_run_id") != run.id or { - s["id"] for s in experiment["inputs"] - } != set(run.authorization["input_ids"]): + if ( + experiment["evidence"].get("flow_run_id") != run.id + or {s["id"] for s in experiment["inputs"]}.issubset(set(run.authorization["input_ids"])) is False + ): raise HTTPException(403, "候选不属于此研究运行的固定输入范围") if "backtest" not in run.authorization["methods"] or any( - c["settings"] != run.authorization["settings"] + c["settings"] + not in ( + run.authorization.get("allowed_settings", [run.authorization["settings"]]) + if experiment["evidence"].get("method") == "settings" + else [run.authorization["settings"]] + ) + or not c.get("input_ids") + or any( + not any( + i["id"] == input_id + and i["scope"] == scope_of(SimulationSettings.model_validate(c["settings"])) + for i in run.authorization["inputs"] + ) + for input_id in c["input_ids"] + ) for c in experiment["candidates"] if c["client_item_id"] in step.output["candidate_ids"] ): diff --git a/backend/app/research/workflows.py b/backend/app/research/workflows.py index 9c749ff..e6e7052 100644 --- a/backend/app/research/workflows.py +++ b/backend/app/research/workflows.py @@ -5,7 +5,7 @@ from collections import defaultdict from fastapi import HTTPException from sqlalchemy import func, select -from ..backtests.contracts import fingerprint +from ..backtests.contracts import SimulationSettings, fingerprint from ..backtests.service import uid from ..models import Account, ResearchFlowRun, ResearchStepRun from .assets import Assets @@ -64,8 +64,12 @@ def validate_graph(graph): if node.type != "summarize" and len(incoming[node.id]) > 1: raise HTTPException(422, "仅汇总节点接受多个上游；其他节点需要唯一输入") allowed = ( - {"prompt"} - if node.type in ("feature", "generate") + {"prompt", "asset_id", "version"} + if node.type == "feature" + else {"asset_id", "version"} + if node.type == "expand" + else {"prompt"} + if node.type == "generate" else {"method"} if node.type == "variant" else {"verdicts"} @@ -76,6 +80,15 @@ def validate_graph(graph): ) if set(node.config) - allowed: raise HTTPException(422, f"节点 {node.id} 包含不支持的配置") + if "asset_id" in node.config or "version" in node.config: + if ( + not isinstance(node.config.get("asset_id"), str) + or not node.config["asset_id"] + or len(node.config["asset_id"]) > 36 + or type(node.config.get("version")) is not int + or node.config["version"] < 1 + ): + raise HTTPException(422, "节点素材引用需要 ID 和正整数版本") if "prompt" in node.config and ( not isinstance(node.config["prompt"], str) or len(node.config["prompt"]) > 10000 ): @@ -84,7 +97,7 @@ def validate_graph(graph): raise HTTPException(422, "未知变体方法") if node.type == "filter" and ( not isinstance(node.config.get("verdicts", ["pass"]), list) - or not set(node.config.get("verdicts", ["pass"])).issubset({"pass", "review", "block"}) + or any(v not in ("pass", "review", "block") for v in node.config.get("verdicts", ["pass"])) ): raise HTTPException(422, "筛选结果必须为 pass/review/block") if node.type == "iterate" and ( @@ -169,8 +182,48 @@ class Workflows: if node.type == "iterate" and node.config["max_rounds"] > body.budget.max_rounds: raise HTTPException(422, "流程迭代上限超过本次授权轮数") experiments = Experiments(self.db) - inputs, _ = await experiments.inputs(body.input_ids, scope_of(body.settings)) + settings_variant = any( + n.type == "variant" and n.config.get("method") == "settings" for n in graph.nodes + ) + if settings_variant and len(body.parent_alpha_ids) != 1: + raise HTTPException(422, "设置变体需要且只能选择一个种子 Alpha") + if any(n.type == "variant" for n in graph.nodes) and not body.parent_alpha_ids: + raise HTTPException(422, "变体节点需要种子 Alpha") + inputs, _ = await experiments.inputs( + body.input_ids, None if settings_variant else scope_of(body.settings) + ) parents = await experiments.parents(body.parent_alpha_ids, []) + allowed_settings = [body.settings.model_dump(mode="json")] + if settings_variant: + base = SimulationSettings.model_validate(parents[0]["settings"]) + for snapshot in inputs: + scope = snapshot["scope"] + target = SimulationSettings.model_validate( + { + **base.model_dump(), + "instrumentType": scope["instrument_type"], + **{key: scope[key] for key in ("region", "universe", "delay")}, + } + ) + errors, _ = await experiments.settings_check(target) + if errors: + raise HTTPException(422, "目标范围：" + "；".join(errors)) + if target.model_dump(mode="json") not in allowed_settings: + allowed_settings.append(target.model_dump(mode="json")) + node_assets = {} + for node in graph.nodes: + if node.config.get("asset_id"): + ref = await Assets(self.db).get( + node.config["asset_id"], + node.config["version"], + "feature" if node.type == "feature" else "template", + ) + if node.type == "feature": + if not set(ref["content"]["input_ids"]).issubset(body.input_ids): + raise HTTPException(422, "特征方案输入超出本次固定范围") + if not ref["content"].get("template"): + raise HTTPException(422, "特征方案需要输出模板") + node_assets[node.id] = ref errors, settings_snapshot = await experiments.settings_check(body.settings) if errors: raise HTTPException(422, "；".join(errors)) @@ -187,6 +240,7 @@ class Workflows: if ( any( node.type == "expand" + and not node.config.get("asset_id") and any( e.target == node.id and next(n for n in graph.nodes if n.id == e.source).type == "input" for e in graph.edges @@ -212,6 +266,8 @@ class Workflows: "template": template, "workflow": asset, "methods": methods, + "node_assets": node_assets, + "allowed_settings": allowed_settings, "settings_snapshot": settings_snapshot, "operators_snapshot": operators_snapshot, "kind": "quantflow" if asset else "pipeline", @@ -272,13 +328,16 @@ class Workflows: } ) - async def list(self, limit=25, offset=0): + async def list(self, limit=25, offset=0, kind=None): + query = select(ResearchFlowRun) + if kind: + query = query.where(ResearchFlowRun.authorization["kind"].as_string() == kind) rows = await self.db.scalars( - select(ResearchFlowRun).order_by(ResearchFlowRun.created_at.desc()).limit(limit).offset(offset) + query.order_by(ResearchFlowRun.created_at.desc()).limit(limit).offset(offset) ) return { "items": [await self.get(row.id) for row in rows], - "total": await self.db.scalar(select(func.count()).select_from(ResearchFlowRun)), + "total": await self.db.scalar(select(func.count()).select_from(query.subquery())), "limit": limit, "offset": offset, } diff --git a/backend/app/research/workspace_tools.py b/backend/app/research/workspace_tools.py index 39008fe..aef966e 100644 --- a/backend/app/research/workspace_tools.py +++ b/backend/app/research/workspace_tools.py @@ -184,6 +184,24 @@ async def list_flows(ctx, args): CAPABILITIES += ( + Capability( + name="search_research_workflows", + schema=AssetQuery, + description="分页查阅 QuantFlow 原生流程与版本，不启动运行。", + label="搜索研究流程", + renderer="research", + effect="query", + handler=lambda ctx, args: Assets(ctx.business.db).list("workflow", **args.model_dump()), + ), + Capability( + name="get_research_workflow", + schema=FixedAssetReference, + description="读取指定研究流程版本及原生节点连接，配合 get_research_run 解释执行产物。", + label="读取流程版本", + renderer="research", + effect="query", + handler=lambda ctx, args: Assets(ctx.business.db).get(args.asset_id, args.version, "workflow"), + ), Capability( name="get_research_run", schema=FlowReference, diff --git a/backend/tests/research_quantflow_postgres.py b/backend/tests/research_quantflow_postgres.py new file mode 100644 index 0000000..169f9e4 --- /dev/null +++ b/backend/tests/research_quantflow_postgres.py @@ -0,0 +1,121 @@ +"""Stage-four PostgreSQL acceptance in dedicated databases only.""" + +import asyncio +import os +import subprocess +from pathlib import Path + +from alembic import command +from alembic.config import Config +from cryptography.fernet import Fernet + +NAME = "wq_research_stage4_test" +RESTORE = "wq_research_restore_stage4" +os.environ.update( + DATABASE_URL=f"postgresql+asyncpg://postgres:research-test-only@127.0.0.1:18436/{NAME}", + ADMIN_PASSWORD="research-acceptance-only", + ENCRYPTION_KEY=Fernet.generate_key().decode(), +) + + +def docker(*args, **kwargs): + return subprocess.run(["docker", "exec", "-i", "wq-research-acceptance-pg", *args], check=True, **kwargs) + + +async def acceptance(): + from unittest.mock import patch + + import httpx + from sqlalchemy import select + + from app.config import Settings + from app.main import create_app + from app.models import TemplateInput + from app.research.workspace_contracts import TemplateSpec + from tests.test_ai import configure + from tests.test_backtests import setup + from tests.test_quantflow import graph, launch, save + from tests.test_research_flows import drive + from tests.test_research_workspace import template + + app = create_app(Settings(_env_file=None, enable_runner=False, public_origin="http://testserver")) + async with app.router.lifespan_context(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app), base_url="http://testserver", headers={"X-WQ-Request": "1"} + ) as client: + assert ( + await client.post( + "/api/v1/auth/login", json={"username": "admin", "password": "research-acceptance-only"} + ) + ).status_code == 200 + # Configure a deterministic model; no provider or real platform network. + from tests.ai_fake import fake_model + + app.state.ai.model_factory = fake_model + await configure(app, client) + platform, lane = await setup(app) + calls = [] + + async def model(ai, context, output_type, revision): + calls.append(context) + value = template() + value["expression"] = f"rank({{field}}) + {len(calls)}" + return TemplateSpec.model_validate(value), { + "model": "fixture", + "revision": revision, + "usage": {"requests": 1}, + } + + async with app.state.sessions() as db: + fixed = await db.scalar(select(TemplateInput)) + body = { + "request_id": "finite-run", + "name": "PG 有限研究", + "input_ids": [fixed.id], + "hypothesis": "排名稳定性", + "settings": {"region": "USA", "universe": "TOP3000", "delay": 1}, + "budget": {"max_rounds": 2, "max_simulations": 4, "max_model_calls": 3}, + "batch_candidates": 2, + } + body["request_id"] = "pg-quantflow" + fixed_template = await save(client, "template", template()) + definition = graph("input", "expand", "backtest", "evaluate", "filter", "iterate") + definition["nodes"][1]["config"] = {"asset_id": fixed_template["id"], "version": 1} + definition["nodes"][4]["config"] = {"verdicts": ["pass", "review", "block"]} + with patch("app.research.runtime.request_model", model): + run = await launch(client, body, definition) + result = await drive(app, client, run["id"], lane, ticks=60) + assert result["status"] == "completed", result + assert ( + result["round"] == 2 + and result["simulations_used"] == 4 + and result["model_calls_used"] == 0 + ) + assert not calls and len(platform.posts) == 2 + await app.state.research.recover() + await app.state.research.advance(run["id"]) + assert len(platform.posts) == 2 + print("PASS PostgreSQL: versioned native graph, filtering, bounded iteration and replay") + + +if __name__ == "__main__": + docker("createdb", "-U", "postgres", NAME) + with Path("/tmp/wq-research-stage3.dump").open("rb") as source: + docker("pg_restore", "-U", "postgres", "-d", NAME, stdin=source) + config = Config("alembic.ini") + command.upgrade(config, "0008") + command.check(config) + asyncio.run(acceptance()) + dump = Path("/tmp/wq-research-stage4.dump") + with dump.open("wb") as output: + docker("pg_dump", "-U", "postgres", "-Fc", NAME, stdout=output) + docker("createdb", "-U", "postgres", RESTORE) + with dump.open("rb") as source: + docker("pg_restore", "-U", "postgres", "-d", RESTORE, stdin=source) + query = "SELECT (SELECT md5(string_agg(row_to_json(t)::text, '' ORDER BY id)) FROM research_flow_runs t),(SELECT md5(string_agg(row_to_json(t)::text, '' ORDER BY id)) FROM research_step_runs t),(SELECT count(*) FROM research_revisions),(SELECT note FROM research WHERE alpha_id='OLD_RESEARCH')" + a = docker("psql", "-U", "postgres", "-d", NAME, "-Atc", query, capture_output=True).stdout + b = docker("psql", "-U", "postgres", "-d", RESTORE, "-Atc", query, capture_output=True).stdout + assert a == b + print( + "PASS PostgreSQL 17: head 0008 unchanged, pg_dump/pg_restore preserves complete run/step snapshots and old research notes" + ) diff --git a/backend/tests/test_quantflow.py b/backend/tests/test_quantflow.py new file mode 100644 index 0000000..4503571 --- /dev/null +++ b/backend/tests/test_quantflow.py @@ -0,0 +1,286 @@ +"""Native graph execution shares immutable inputs, artifacts and budgeted backtests.""" + +import copy + +import pytest + +from app.alphas import upsert_alpha +from app.models import Alpha, CatalogResource +from app.research.workspace_contracts import FeatureSpec +from tests.conftest import alpha +from tests.test_catalog import SCOPE, prepare, sync +from tests.test_research_flows import begin, drive, flow_setup, get +from tests.test_research_workspace import catalog, research_input, template + +__all__ = ["flow_setup", "catalog", "research_input"] + + +def graph(*kinds): + return { + "name": "原生节点研究", + "nodes": [ + { + "id": f"n{i}", + "type": kind, + "label": kind, + "config": {"max_rounds": 2} if kind == "iterate" else {}, + } + for i, kind in enumerate(kinds) + ], + "edges": [{"source": f"n{i}", "target": f"n{i + 1}"} for i in range(len(kinds) - 1)], + } + + +async def save(client, kind, content): + r = await client.post("/api/v1/research/assets", json={"kind": kind, "content": content}) + assert r.status_code == 201, r.text + return r.json() + + +async def launch(client, body, definition): + asset = await save(client, "workflow", definition) + return await begin(client, {**body, "workflow_id": asset["id"], "workflow_version": asset["version"]}) + + +@pytest.mark.parametrize( + "case", ["cycle", "type", "duplicate", "disconnected", "loop", "config", "filter", "ref"] +) +async def test_invalid_graphs_rejected_before_save(logged_in, case): + value = graph("input", "generate", "expand", "backtest", "evaluate") + if case == "cycle": + value["nodes"] += [{"id": "loop_a", "type": "generate"}, {"id": "loop_b", "type": "expand"}] + value["edges"] += [{"source": "loop_a", "target": "loop_b"}, {"source": "loop_b", "target": "loop_a"}] + if case == "type": + value["edges"][0]["target"] = "n3" + if case == "duplicate": + value["edges"].append(value["edges"][0]) + if case == "disconnected": + value["edges"].pop() + if case == "loop": + value["nodes"].append({"id": "iterate", "type": "iterate", "config": {"max_rounds": True}}) + value["edges"].append({"source": "n4", "target": "iterate"}) + if case == "config": + value["nodes"][1]["config"] = {"command": "bash"} + if case == "filter": + value["nodes"].append({"id": "filter", "type": "filter", "config": {"verdicts": [{}]}}) + value["edges"].append({"source": "n4", "target": "filter"}) + if case == "ref": + value["nodes"][2]["config"] = {"asset_id": "x", "version": True} + for url in ("/api/v1/research/flows/validate", "/api/v1/research/assets"): + r = await logged_in.post( + url, json=value if url.endswith("validate") else {"kind": "workflow", "content": value} + ) + assert r.status_code == 422, r.text + + +async def test_versioned_template_branch_join_and_replayed_events(app, logged_in, flow_setup): + body, platform, lane, calls = flow_setup + fixed = await save(logged_in, "template", template()) + definition = graph("input", "expand", "backtest", "evaluate", "condition") + definition["nodes"][1]["config"] = {"asset_id": fixed["id"], "version": 1} + for verdict in ("pass", "review", "block"): + definition["nodes"].append({"id": verdict, "type": "summarize"}) + definition["edges"].append({"source": "n4", "target": verdict, "branch": verdict}) + # A shared join receives the condition's evidence as well as direct evaluation. + definition["nodes"].append({"id": "summary", "type": "summarize"}) + definition["edges"] += [{"source": "n4", "target": "summary"}, {"source": "n3", "target": "summary"}] + run = await launch(logged_in, body, definition) + workflow = run["authorization"]["workflow"] + modified = copy.deepcopy(definition) + modified["nodes"][1]["config"]["version"] = 2 + assert ( + await logged_in.put( + f"/api/v1/research/assets/{workflow['id']}", + json={"kind": "workflow", "version": 1, "content": modified}, + ) + ).status_code == 200 + value = template() + value["expression"] = "unknown({field})" + assert ( + await logged_in.put( + f"/api/v1/research/assets/{fixed['id']}", + json={"kind": "template", "version": 1, "content": value}, + ) + ).status_code == 200 + final = await drive(app, logged_in, run["id"], lane) + assert final["status"] == "completed", final + assert final["model_calls_used"] == 0 and not calls and final["simulations_used"] == 2 + branches = [s for s in final["steps"] if s["node_id"] in ("pass", "review", "block")] + verdict = next(s for s in final["steps"] if s["node_id"] == "n3")["output"]["verdict"] + assert [s["node_id"] for s in branches if s["status"] == "completed"] == [verdict] + assert sum(s["status"] == "skipped" for s in branches) == 2 + assert len(next(s for s in final["steps"] if s["node_id"] == "summary")["output"]["artifacts"]) == 2 + assert final["authorization"]["node_assets"]["n1"]["version"] == 1 + for _ in range(3): + await app.state.research.advance(run["id"]) + assert len(platform.posts) == 1 + assert (await logged_in.get("/api/v1/research/flows/runs?kind=quantflow")).json()["total"] == 1 + assert (await logged_in.get("/api/v1/research/flows/runs?kind=pipeline")).json()["total"] == 0 + + +async def test_feature_reference_conversion_and_bounded_loop(app, logged_in, flow_setup): + body, _, lane, calls = flow_setup + feature = await save( + logged_in, + "feature", + { + "name": "字段方案", + "hypothesis": "排名稳定性", + "input_ids": body["input_ids"], + "steps": [{"name": "排名", "rationale": "截面比较", "expression": "rank(TEST_FIN_001)"}], + "template": template(), + }, + ) + definition = graph("input", "feature", "expand", "backtest", "evaluate", "filter", "iterate") + definition["nodes"][1]["config"] = {"asset_id": feature["id"], "version": 1} + definition["nodes"][5]["config"] = {"verdicts": ["pass", "review", "block"]} + run = await launch(logged_in, body, definition) + final = await drive(app, logged_in, run["id"], lane, ticks=60) + assert final["status"] == "completed", final + assert final["round"] == 2 and final["simulations_used"] == 4 and not calls + outputs = [s["output"] for s in final["steps"] if s["node_id"] == "n1"] + assert all(o["feature"]["id"] == feature["id"] for o in outputs) + template_asset = (await logged_in.get(f"/api/v1/research/assets/{outputs[0]['template']['id']}")).json() + assert template_asset["provenance"]["feature"]["version"] == 1 + + +async def test_native_feature_model_and_structure_variant(app, logged_in, flow_setup, monkeypatch): + body, _, lane, calls = flow_setup + + async def feature_model(ai, context, output_type, revision): + assert output_type is FeatureSpec + calls.append(context) + return FeatureSpec.model_validate( + { + "name": "模型特征", + "hypothesis": "排名", + "input_ids": body["input_ids"], + "steps": [{"name": "排名", "rationale": "比较", "expression": "rank(TEST_FIN_001)"}], + "template": template(), + } + ), {"model": "fixture", "revision": revision} + + with monkeypatch.context() as patch: + patch.setattr("app.research.runtime.request_model", feature_model) + run = await launch(logged_in, body, graph("input", "feature", "expand", "backtest", "evaluate")) + final = await drive(app, logged_in, run["id"], lane) + assert final["status"] == "completed" and final["model_calls_used"] == 1, final + assert final["steps"][1]["output"]["feature"] + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed", regular={"code": "rank(TEST_FIN_001)"})) + run = await launch( + logged_in, + {**body, "request_id": "variant", "parent_alpha_ids": ["seed"]}, + graph("input", "variant", "backtest", "evaluate"), + ) + final = await drive(app, logged_in, run["id"], lane) + assert final["status"] == "completed" and final["model_calls_used"] == 1, final + experiment = ( + await logged_in.get(f"/api/v1/research/experiments/{final['steps'][1]['output']['experiment_id']}") + ).json() + assert experiment["parents"][0]["id"] == "seed" and experiment["kind"] == "quantflow" + + +async def test_settings_variant_uses_frozen_seed_and_target_scope(app, logged_in, research_input, catalog): + from tests.test_backtests import setup + + body = { + "request_id": "scope-run", + "name": "范围变体", + "input_ids": [research_input["id"]], + "hypothesis": "跨股票池比较", + "settings": {"region": "USA", "universe": "TOP3000", "delay": 1}, + "budget": {"max_rounds": 1, "max_simulations": 2, "max_model_calls": 1}, + } + target = {**SCOPE, "universe": "TOP1000"} + await sync(catalog, scope=target) + version = (await sync(catalog, "TEST_FIN", scope=target))["id"] + fixed = (await prepare(logged_in, version, scope=target)).json() + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed", regular={"code": "x = TEST_FIN_001; rank(x)"})) + metadata = await db.get(CatalogResource, "settings") + metadata.content = { + "items": metadata.content["items"] + [{**target, "neutralizations": ["INDUSTRY"]}] + } + _, lane = await setup(app) + definition = graph("input", "variant", "backtest", "evaluate") + definition["nodes"][1]["config"] = {"method": "settings"} + run = await launch( + logged_in, + {**body, "input_ids": body["input_ids"] + [fixed["id"]], "parent_alpha_ids": ["seed"]}, + definition, + ) + async with app.state.sessions.begin() as db: + seed = await db.get(Alpha, "seed") + seed.expression = "rank(unknown)" + final = await drive(app, logged_in, run["id"], lane) + assert ( + final["status"] == "completed" and final["simulations_used"] == 1 and final["model_calls_used"] == 0 + ), final + experiment = ( + await logged_in.get(f"/api/v1/research/experiments/{final['steps'][1]['output']['experiment_id']}") + ).json() + c = experiment["candidates"][0] + assert c["expression"] == "x = TEST_FIN_001; rank(x)" and c["input_ids"] == [fixed["id"]] + assert c["settings"]["universe"] == "TOP1000" and len(experiment["inputs"]) == 2 + + +async def test_empty_filter_skips_downstream_simulation(app, logged_in, flow_setup): + body, platform, lane, _ = flow_setup + definition = graph( + "input", "generate", "expand", "backtest", "evaluate", "filter", "backtest", "evaluate" + ) + definition["nodes"][5]["config"] = {"verdicts": []} + run = await launch(logged_in, body, definition) + final = await drive(app, logged_in, run["id"], lane) + assert final["status"] == "completed" and len(platform.posts) == 1, final + assert [s["status"] for s in final["steps"][-3:]] == ["skipped"] * 3 + + +@pytest.mark.parametrize("action", ["pause", "stop", "account"]) +async def test_collected_variant_output_waits_for_authorized_next_tick(app, logged_in, flow_setup, action): + from sqlalchemy import func, select + + from app.models import Account, ResearchExperiment + from app.research.workspace_contracts import TemplateSpec + + body, platform, lane, _ = flow_setup + async with app.state.sessions.begin() as db: + await upsert_alpha(db, alpha("seed", regular={"code": "rank(TEST_FIN_001)"})) + run = await launch( + logged_in, {**body, "parent_alpha_ids": ["seed"]}, graph("input", "variant", "backtest", "evaluate") + ) + await app.state.research.advance(run["id"]) + work = await app.state.research.prepare(run["id"]) + current = await get(logged_in, run["id"]) + if action == "account": + async with app.state.sessions.begin() as db: + account = await db.get(Account, 1) + account.wq_user_id = "OTHER_ACCOUNT" + else: + assert ( + await logged_in.post( + f"/api/v1/research/flows/runs/{run['id']}/control", + json={"action": action, "version": current["version"]}, + ) + ).status_code == 200 + await app.state.research.finish_model( + run["id"], work[0], TemplateSpec.model_validate(template()), {"model": "fixture"} + ) + await app.state.research.advance(run["id"]) + current = await get(logged_in, run["id"]) + assert current["steps"][1]["status"] == "generated" and current["model_calls_used"] == 1 + assert current["status"] == {"pause": "paused", "stop": "stopped", "account": "interrupted"}[action] + async with app.state.sessions() as db: + assert await db.scalar(select(func.count()).select_from(ResearchExperiment)) == 0 + assert not platform.posts + if action == "pause": + assert ( + await logged_in.post( + f"/api/v1/research/flows/runs/{run['id']}/control", + json={"action": "resume", "version": current["version"]}, + ) + ).status_code == 200 + await app.state.research.recover() + final = await drive(app, logged_in, run["id"], lane) + assert final["status"] == "completed" and final["model_calls_used"] == 1 diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index f6d912e..0af5200 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -12,6 +12,7 @@ import { } from "@douyinfe/semi-ui-19"; import zhCN from "@douyinfe/semi-ui-19/lib/es/locale/source/zh_CN"; import { IconSetting, IconComment, IconHistory } from "@douyinfe/semi-icons"; +import { QuantFlowPage } from "./research/QuantFlowPage"; import { PipelinePage } from "./research/PipelinePage"; import { FeaturesPage } from "./research/FeaturesPage"; import { ResearchWorkspace } from "./research/ResearchWorkspace"; @@ -52,9 +53,14 @@ export default function App() { }); useEffect(() => { if ( - ["operators", "templates", "variants", "features", "pipeline"].includes( - page, - ) + [ + "operators", + "templates", + "variants", + "features", + "pipeline", + "quantflow", + ].includes(page) ) setVisitedResearch((old) => (old.includes(page) ? old : [...old, page])); }, [page]); @@ -376,6 +382,15 @@ export default function App() { /> )} + {visitedResearch.includes("quantflow") && ( + + )} {visitedResearch.includes("pipeline") && (