From 222595d6c4daa5bf0687eda404ce67df6c34fb99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Sun, 30 Aug 2026 20:02:15 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=B7=A5=E4=BD=9C=E6=B5=81=E7=94=BB?= =?UTF-8?q?=E5=B8=83=E6=94=AF=E6=8C=81=E8=8D=89=E7=A8=BF=E8=AF=95=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=EF=BC=88target=3Ddraft=20=E6=89=A7=E8=A1=8C=E5=BD=93?= =?UTF-8?q?=E5=89=8D=E8=8D=89=E7=A8=BF=E5=9B=BE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/gateway/routers/workflow_runs.py | 54 +++++++---- .../gateway/routers/workflows_coze_compat.py | 95 ++++++++++++++++++- .../backend/app/gateway/workflow_executor.py | 9 ++ .../app/gateway/workflow_resource_checks.py | 14 ++- .../deerflow/workflows/nodes/agents.py | 16 +++- .../tests/test_workflow_definitions.py | 1 + .../backend/tests/test_workflow_runs.py | 41 ++++++++ 7 files changed, 198 insertions(+), 32 deletions(-) diff --git a/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py b/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py index 0d71e4b2..68b38ee1 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py +++ b/offline-backend-20260512/backend/app/gateway/routers/workflow_runs.py @@ -149,26 +149,36 @@ def _public_run(run: dict[str, Any]) -> dict[str, Any]: async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> dict[str, Any]: _require_enabled() user_id = await get_current_user(request) - if str(body.target or "").strip().lower() == "draft": - # Never silently fall back to the published version when the client - # thinks it is exercising the draft — draft snapshots need the explicit - # test-run API (not built yet). - raise HTTPException( - status_code=422, - detail={ - "code": "WORKFLOW_DRAFT_RUN_UNSUPPORTED", - "message": "暂不支持运行草稿，请先发布或指定 versionId", - }, - ) definitions = _definitions(request) - definition = await definitions.get_definition(workflow_id, include_draft=False) + is_draft_run = str(body.target or "").strip().lower() == "draft" + + if is_draft_run: + # Test-run semantics: execute the CURRENT draft graph (what the canvas + # shows right now), versioned as the literal "draft". The compiled + # execution graph travels in run.context (never the canvas document), + # and the executor prefers it over any published version. + definition = await definitions.get_definition(workflow_id, include_draft=True) + else: + definition = await definitions.get_definition(workflow_id, include_draft=False) if definition is None: raise HTTPException(status_code=404, detail="工作流不存在") if definition.get("owner_id") != user_id and not await _is_admin(request): raise HTTPException(status_code=403, detail={"code": "WORKFLOW_FORBIDDEN", "message": "无权运行该工作流"}) + draft_graph_raw: dict[str, Any] | None = None + if is_draft_run: + draft_graph_raw = definition.get("draft_graph") or {} + if not draft_graph_raw.get("nodes"): + raise HTTPException( + status_code=400, + detail={ + "code": "WORKFLOW_DRAFT_EMPTY", + "message": "草稿还没有可运行的执行图，请先在画布保存一次", + }, + ) + version_id = body.version_id - if not version_id: + if not version_id and not is_draft_run: versions = await definitions.list_versions(workflow_id) if not versions: raise HTTPException( @@ -176,11 +186,14 @@ async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> d detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "工作流尚未发布任何版本"}, ) version_id = versions[0]["id"] - version = await definitions.get_version(version_id) - if version is None or version.get("workflow_id") != workflow_id: - raise HTTPException(status_code=404, detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "版本不存在"}) + if is_draft_run: + version_id = "draft" + else: + version = await definitions.get_version(version_id) + if version is None or version.get("workflow_id") != workflow_id: + raise HTTPException(status_code=404, detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "版本不存在"}) - graph = WorkflowGraph.model_validate(version.get("graph") or {}) + graph = WorkflowGraph.model_validate(draft_graph_raw if is_draft_run else (version.get("graph") or {})) inputs = body.resolved_inputs() missing = [key for key in (graph.input_schema.get("required") or []) if inputs.get(str(key)) in (None, "")] if missing: @@ -218,6 +231,11 @@ async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> d # Scope the idempotency key per workflow so two workflows can share a client key. key = f"{workflow_id}:{body.idempotency_key}" if body.idempotency_key else None + context: dict[str, Any] = {"depth": 0} + if is_draft_run: + # The draft snapshot rides with the run; the executor prefers it over + # the published version. Execution graph only — never the canvas doc. + context["draftGraph"] = draft_graph_raw run, created = await runs.create_run( { "workflow_id": workflow_id, @@ -225,7 +243,7 @@ async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> d "owner_id": user_id, "input": inputs, "idempotency_key": key, - "context": {"depth": 0}, + "context": context, } ) if created: diff --git a/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py b/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py index 255016b6..920ea657 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py +++ b/offline-backend-20260512/backend/app/gateway/routers/workflows_coze_compat.py @@ -278,10 +278,46 @@ async def coze_playground_space_save(request: Request) -> dict[str, Any]: return _coze_ok({"id": _DEFAULT_SPACE["id"], "check_not_pass": False}) +def _coze_model_type(name: str) -> int: + """Stable positive int32 from the model name. + + The Coze LLM form stores the picked model as ``modelType`` (number) plus + ``modleName`` (string, upstream typo) — the compiler reads the string, so + the number only needs to be stable within a deployment, not reversible. + """ + digest = 5381 + for char in name: + digest = ((digest * 33) + ord(char)) & 0xFFFFFFFF + return digest & 0x7FFFFFFF + + @chrome_router.api_route("/api/bot/get_type_list", methods=["GET", "POST"]) async def coze_model_type_list(request: Request) -> dict[str, Any]: + """LLM form model list → real DeerFlow models (names only, no secrets). + + The Coze ``Model`` shape is trimmed to what the workflow form consumes: + ``model_type`` (stable number), ``name`` (the DeerFlow model name, stored + verbatim into the node so the runtime can override the model), and empty + ability/params maps — ``getLLMModels`` fills in response_format defaults + client-side. Parameter tuning panels stay inert: model tuning happens in + DeerFlow, the workflow node only picks which model to call. + """ await _require_studio_user(request) - return _coze_ok({"model_list": []}) + config = get_app_config() + configured = list(getattr(config, 'models', None) or []) + model_list = [ + { + "model_type": _coze_model_type(model.name), + "name": model.name, + "display_name": model.display_name or model.name, + "description": "", + "model_ability": {}, + "model_params": [], + "is_custom": False, + } + for model in configured + ] + return _coze_ok({"model_list": model_list}) @router.post("/create") @@ -366,3 +402,60 @@ async def coze_node_template_list(request: Request) -> dict[str, Any]: "plugin_category_list": [], }, } + + +@router.post("/workflow_references") +async def coze_workflow_references(request: Request) -> dict[str, Any]: + """Which workflows reference the current one via subworkflow nodes. + + Honest empty answer: the standalone page's subworkflow form is not wired + yet, so no canvas can reference another — an empty list is the truthful + state, not a placeholder for missing data. + """ + _require_enabled() + await get_current_user(request) + return {"code": 0, "msg": "success", "data": {"workflow_list": []}} + + +@router.post("/nodeDebug") +async def coze_node_debug(request: Request) -> dict[str, Any]: + """Single-node debug: NOT wired to the DeerFlow executor. + + Deliberately a business error (Coze envelope code != 0) so the playground + surfaces an explicit message instead of pretending the debug succeeded. + """ + _require_enabled() + await get_current_user(request) + return { + "code": 4003101, + "msg": "节点单点调试暂未接入 DeerFlow 执行器，请使用画布下方「试运行」运行整个工作流", + } + + +@router.post("/llm_fc_setting_detail") +@router.post("/llm_fc_setting_merged") +async def coze_llm_fc_setting(request: Request) -> dict[str, Any]: + """LLM function-call settings detail/merge for plugin/workflow lists. + + No Coze plugins are wired into the canvas, so every detail map is empty. + """ + _require_enabled() + await get_current_user(request) + return { + "code": 0, + "msg": "success", + "data": { + "plugin_detail_map": {}, + "plugin_api_detail_map": {}, + "workflow_detail_map": {}, + "dataset_detail_map": {}, + "knowledge_detail_map": {}, + }, + } + + +@chrome_router.post("/api/devops/debugger/v1/coze/testcase/casedata/mget") +async def coze_testcase_casedata_mget(request: Request) -> dict[str, Any]: + """Playground test-case fetch: no Coze test cases exist for this page.""" + await _require_studio_user(request) + return {"code": 0, "msg": "success", "data": {"case_data_list": [], "total": 0}} diff --git a/offline-backend-20260512/backend/app/gateway/workflow_executor.py b/offline-backend-20260512/backend/app/gateway/workflow_executor.py index 6b1e42c4..21e52e9f 100644 --- a/offline-backend-20260512/backend/app/gateway/workflow_executor.py +++ b/offline-backend-20260512/backend/app/gateway/workflow_executor.py @@ -274,6 +274,15 @@ class WorkflowRunExecutor: ) async def _load_graph(self, run: dict[str, Any]) -> WorkflowGraph | None: + # Draft test-runs carry their execution graph in run.context (version id + # is the literal "draft", no published row exists for it). + draft_graph = (run.get("context") or {}).get("draftGraph") + if draft_graph: + try: + return WorkflowGraph.model_validate(draft_graph) + except Exception: # noqa: BLE001 + logger.warning("draft run %s has an invalid graph", run["id"], exc_info=True) + return None version = await self._workflows.get_version(run["workflow_version_id"]) if version is None or version.get("workflow_id") != run["workflow_id"]: return None diff --git a/offline-backend-20260512/backend/app/gateway/workflow_resource_checks.py b/offline-backend-20260512/backend/app/gateway/workflow_resource_checks.py index 4394dbd4..f4481e73 100644 --- a/offline-backend-20260512/backend/app/gateway/workflow_resource_checks.py +++ b/offline-backend-20260512/backend/app/gateway/workflow_resource_checks.py @@ -117,10 +117,10 @@ async def build_resource_catalog(request: Request, *, user_id: str) -> ResourceC def _check_agent(config: dict[str, Any], catalog: ResourceCatalog) -> dict[str, Any] | None: - agent_id = str(config.get("agentId") or config.get("agent_id") or "") - if not agent_id: - return _issue("未选择智能体", "config.agentId") - if agent_id not in catalog.agent_ids: + # Empty agentId means the runtime's built-in default agent — a real, + # executable fallback, not missing configuration. + agent_id = str(config.get("agentId") or config.get("agent_id") or "") or "default" + if agent_id != "default" and agent_id not in catalog.agent_ids: return _issue("未选择可访问的智能体", "config.agentId") if not str(config.get("promptTemplate") or config.get("prompt_template") or "").strip(): return _issue("智能体节点的提示词不能为空", "config.promptTemplate") @@ -138,10 +138,8 @@ def _check_skill(config: dict[str, Any], catalog: ResourceCatalog) -> dict[str, mode = str(config.get("mode") or "agent_skill") if mode != "agent_skill": return _issue("callable_skill 模式暂未开放执行，请改用 agent_skill", "config.mode") - agent_id = str(config.get("agentId") or config.get("agent_id") or "") - if not agent_id: - return _issue("技能节点未选择智能体", "config.agentId") - if agent_id not in catalog.agent_ids: + agent_id = str(config.get("agentId") or config.get("agent_id") or "") or "default" + if agent_id != "default" and agent_id not in catalog.agent_ids: return _issue("技能节点引用的智能体不可访问", "config.agentId") skill_names = [str(s) for s in (config.get("skillNames") or config.get("skill_names") or [])] if not skill_names: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py b/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py index 42e4504e..a4fa0da9 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/workflows/nodes/agents.py @@ -131,6 +131,7 @@ class _AgentInvocation: response_schema: dict[str, Any], thread_mode: str, failure_code: str, + model_name: str = "", ) -> NodeResult: run_agent = require_dep(ctx.deps.run_agent, node=node, what="智能体执行") coalescer = _DeltaCoalescer(ctx, node.id) @@ -176,6 +177,7 @@ class _AgentInvocation: prompt=prompt, skill_names=skill_names, thread_mode=thread_mode, + model_name=model_name or None, cancel=ctx.cancel, on_delta=on_delta, on_tool=on_tool, @@ -251,9 +253,10 @@ class _AgentInvocation: class AgentNodeExecutor(_AgentInvocation): async def execute(self, node: WorkflowNode, ctx: RunContext) -> NodeResult: - agent_id = str(node.config.get("agentId") or node.config.get("agent_id") or "") - if not agent_id: - raise WorkflowError("WORKFLOW_AGENT_FAILED", "智能体节点未选择智能体", node_id=node.id) + # Empty agentId falls back to the runtime's built-in default agent — + # canvas nodes may only pick a model and a prompt. + agent_id = str(node.config.get("agentId") or node.config.get("agent_id") or "") or "default" + model_name = str(node.config.get("modelName") or node.config.get("model_name") or "") template = str(node.config.get("promptTemplate") or node.config.get("prompt_template") or "") prompt = render_template(template, ctx.state()).strip() if not prompt: @@ -264,6 +267,7 @@ class AgentNodeExecutor(_AgentInvocation): ctx, agent_id=agent_id, prompt=prompt, + model_name=model_name, skill_names=[str(s) for s in (node.config.get("skillNames") or node.config.get("skill_names") or [])], response_mode=str(node.config.get("responseMode") or node.config.get("response_mode") or "text"), response_schema=schema if isinstance(schema, dict) else {}, @@ -285,9 +289,10 @@ class SkillNodeExecutor(_AgentInvocation): node_id=node.id, details={"mode": mode}, ) - agent_id = str(node.config.get("agentId") or node.config.get("agent_id") or "") + agent_id = str(node.config.get("agentId") or node.config.get("agent_id") or "") or "default" + model_name = str(node.config.get("modelName") or node.config.get("model_name") or "") skill_names = [str(s) for s in (node.config.get("skillNames") or node.config.get("skill_names") or [])] - if not agent_id or not skill_names: + if not skill_names: raise WorkflowError("WORKFLOW_SKILL_FAILED", "技能节点需要同时指定智能体与技能", node_id=node.id) template = str(node.config.get("promptTemplate") or node.config.get("prompt_template") or "") prompt = render_template(template, ctx.state()).strip() or f"请使用技能 {skill_names[0]} 完成任务。" @@ -296,6 +301,7 @@ class SkillNodeExecutor(_AgentInvocation): ctx, agent_id=agent_id, prompt=prompt, + model_name=model_name, skill_names=skill_names, response_mode="text", response_schema={}, diff --git a/offline-backend-20260512/backend/tests/test_workflow_definitions.py b/offline-backend-20260512/backend/tests/test_workflow_definitions.py index 4f924a25..52aa5cc9 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_definitions.py +++ b/offline-backend-20260512/backend/tests/test_workflow_definitions.py @@ -107,6 +107,7 @@ def test_workflow_http_api_and_coze_compat(monkeypatch: pytest.MonkeyPatch) -> N class _Cfg: workflows = WorkflowConfig() + models = [] monkeypatch.setattr(workflows, "get_app_config", lambda: _Cfg()) monkeypatch.setattr(workflow_resources, "get_app_config", lambda: _Cfg()) diff --git a/offline-backend-20260512/backend/tests/test_workflow_runs.py b/offline-backend-20260512/backend/tests/test_workflow_runs.py index c7a40c43..2080f350 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_runs.py +++ b/offline-backend-20260512/backend/tests/test_workflow_runs.py @@ -705,3 +705,44 @@ async def test_embed_ticket_round_trip(harness: Harness, monkeypatch: pytest.Mon replay = await harness.client.post("/api/workflows/embed/verify", json={"ticket": ticket}) assert replay.status_code == 401 + + +@pytest.mark.asyncio +async def test_draft_run_executes_current_draft_without_publishing(harness: Harness) -> None: + """target=draft runs the CURRENT draft graph (version id "draft"), no publish needed.""" + store = harness.app.state.workflow_store + definition = await store.create_definition({"name": "draft-run", "owner_id": "user-a", "draft_graph": REPORT_GRAPH}) + + response = await harness.client.post( + f"/api/workflows/{definition['id']}/runs", + json={"inputs": {"topic": "储能"}, "target": "draft", "idempotencyKey": "draft-1"}, + ) + assert response.status_code == 200, response.text + run = response.json() + assert run["versionId"] == "draft" + assert run["status"] in ("queued", "running") + + await harness.drive() + + events = await harness.events(run["runId"]) + types = [event["event"] for event in events] + assert "run.started" in types + assert "run.completed" in types + final = await harness.client.get(f"/api/workflows/runs/{run['runId']}") + assert final.status_code == 200 + assert final.json()["status"] == "completed" + + +@pytest.mark.asyncio +async def test_draft_run_without_graph_is_rejected(harness: Harness) -> None: + store = harness.app.state.workflow_store + definition = await store.create_definition( + {"name": "empty-draft", "owner_id": "user-a", "draft_graph": {"nodes": [], "edges": []}} + ) + + response = await harness.client.post( + f"/api/workflows/{definition['id']}/runs", + json={"inputs": {}, "target": "draft"}, + ) + assert response.status_code == 400 + assert response.json()["detail"]["code"] == "WORKFLOW_DRAFT_EMPTY"