feat: 工作流画布支持草稿试运行(target=draft 执行当前草稿图)

This commit is contained in:
董宏凯 committed 2026-08-30 20:02:15 +08:00
1 parent 3384f2e51c
commit 222595d6c4
7 files changed
+194 -28

No files matched your search

@@ -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]: async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> dict[str, Any]:
_require_enabled() _require_enabled()
user_id = await get_current_user(request) 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) definitions = _definitions(request)
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) definition = await definitions.get_definition(workflow_id, include_draft=False)
if definition is None: if definition is None:
raise HTTPException(status_code=404, detail="工作流不存在") raise HTTPException(status_code=404, detail="工作流不存在")
if definition.get("owner_id") != user_id and not await _is_admin(request): if definition.get("owner_id") != user_id and not await _is_admin(request):
raise HTTPException(status_code=403, detail={"code": "WORKFLOW_FORBIDDEN", "message": "无权运行该工作流"}) 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 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) versions = await definitions.list_versions(workflow_id)
if not versions: if not versions:
raise HTTPException( 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": "工作流尚未发布任何版本"}, detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "工作流尚未发布任何版本"},
) )
version_id = versions[0]["id"] version_id = versions[0]["id"]
if is_draft_run:
version_id = "draft"
else:
version = await definitions.get_version(version_id) version = await definitions.get_version(version_id)
if version is None or version.get("workflow_id") != workflow_id: if version is None or version.get("workflow_id") != workflow_id:
raise HTTPException(status_code=404, detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "版本不存在"}) 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() inputs = body.resolved_inputs()
missing = [key for key in (graph.input_schema.get("required") or []) if inputs.get(str(key)) in (None, "")] missing = [key for key in (graph.input_schema.get("required") or []) if inputs.get(str(key)) in (None, "")]
if missing: 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. # 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 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( run, created = await runs.create_run(
{ {
"workflow_id": workflow_id, "workflow_id": workflow_id,
@@ -225,7 +243,7 @@ async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> d
"owner_id": user_id, "owner_id": user_id,
"input": inputs, "input": inputs,
"idempotency_key": key, "idempotency_key": key,
"context": {"depth": 0}, "context": context,
} }
) )
if created: if created:
@@ -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}) 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"]) @chrome_router.api_route("/api/bot/get_type_list", methods=["GET", "POST"])
async def coze_model_type_list(request: Request) -> dict[str, Any]: 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) 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") @router.post("/create")
@@ -366,3 +402,60 @@ async def coze_node_template_list(request: Request) -> dict[str, Any]:
"plugin_category_list": [], "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}}
@@ -274,6 +274,15 @@ class WorkflowRunExecutor:
) )
async def _load_graph(self, run: dict[str, Any]) -> WorkflowGraph | None: 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"]) version = await self._workflows.get_version(run["workflow_version_id"])
if version is None or version.get("workflow_id") != run["workflow_id"]: if version is None or version.get("workflow_id") != run["workflow_id"]:
return None return None
@@ -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: 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 "") # Empty agentId means the runtime's built-in default agent — a real,
if not agent_id: # executable fallback, not missing configuration.
return _issue("未选择智能体", "config.agentId") agent_id = str(config.get("agentId") or config.get("agent_id") or "") or "default"
if agent_id not in catalog.agent_ids: if agent_id != "default" and agent_id not in catalog.agent_ids:
return _issue("未选择可访问的智能体", "config.agentId") return _issue("未选择可访问的智能体", "config.agentId")
if not str(config.get("promptTemplate") or config.get("prompt_template") or "").strip(): if not str(config.get("promptTemplate") or config.get("prompt_template") or "").strip():
return _issue("智能体节点的提示词不能为空", "config.promptTemplate") 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") mode = str(config.get("mode") or "agent_skill")
if mode != "agent_skill": if mode != "agent_skill":
return _issue("callable_skill 模式暂未开放执行,请改用 agent_skill", "config.mode") return _issue("callable_skill 模式暂未开放执行,请改用 agent_skill", "config.mode")
agent_id = str(config.get("agentId") or config.get("agent_id") or "") agent_id = str(config.get("agentId") or config.get("agent_id") or "") or "default"
if not agent_id: if agent_id != "default" and agent_id not in catalog.agent_ids:
return _issue("技能节点未选择智能体", "config.agentId")
if agent_id not in catalog.agent_ids:
return _issue("技能节点引用的智能体不可访问", "config.agentId") return _issue("技能节点引用的智能体不可访问", "config.agentId")
skill_names = [str(s) for s in (config.get("skillNames") or config.get("skill_names") or [])] skill_names = [str(s) for s in (config.get("skillNames") or config.get("skill_names") or [])]
if not skill_names: if not skill_names:
@@ -131,6 +131,7 @@ class _AgentInvocation:
response_schema: dict[str, Any], response_schema: dict[str, Any],
thread_mode: str, thread_mode: str,
failure_code: str, failure_code: str,
model_name: str = "",
) -> NodeResult: ) -> NodeResult:
run_agent = require_dep(ctx.deps.run_agent, node=node, what="智能体执行") run_agent = require_dep(ctx.deps.run_agent, node=node, what="智能体执行")
coalescer = _DeltaCoalescer(ctx, node.id) coalescer = _DeltaCoalescer(ctx, node.id)
@@ -176,6 +177,7 @@ class _AgentInvocation:
prompt=prompt, prompt=prompt,
skill_names=skill_names, skill_names=skill_names,
thread_mode=thread_mode, thread_mode=thread_mode,
model_name=model_name or None,
cancel=ctx.cancel, cancel=ctx.cancel,
on_delta=on_delta, on_delta=on_delta,
on_tool=on_tool, on_tool=on_tool,
@@ -251,9 +253,10 @@ class _AgentInvocation:
class AgentNodeExecutor(_AgentInvocation): class AgentNodeExecutor(_AgentInvocation):
async def execute(self, node: WorkflowNode, ctx: RunContext) -> NodeResult: async def execute(self, node: WorkflowNode, ctx: RunContext) -> NodeResult:
agent_id = str(node.config.get("agentId") or node.config.get("agent_id") or "") # Empty agentId falls back to the runtime's built-in default agent —
if not agent_id: # canvas nodes may only pick a model and a prompt.
raise WorkflowError("WORKFLOW_AGENT_FAILED", "智能体节点未选择智能体", node_id=node.id) 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 "") template = str(node.config.get("promptTemplate") or node.config.get("prompt_template") or "")
prompt = render_template(template, ctx.state()).strip() prompt = render_template(template, ctx.state()).strip()
if not prompt: if not prompt:
@@ -264,6 +267,7 @@ class AgentNodeExecutor(_AgentInvocation):
ctx, ctx,
agent_id=agent_id, agent_id=agent_id,
prompt=prompt, prompt=prompt,
model_name=model_name,
skill_names=[str(s) for s in (node.config.get("skillNames") or node.config.get("skill_names") or [])], 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_mode=str(node.config.get("responseMode") or node.config.get("response_mode") or "text"),
response_schema=schema if isinstance(schema, dict) else {}, response_schema=schema if isinstance(schema, dict) else {},
@@ -285,9 +289,10 @@ class SkillNodeExecutor(_AgentInvocation):
node_id=node.id, node_id=node.id,
details={"mode": mode}, 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 [])] 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) raise WorkflowError("WORKFLOW_SKILL_FAILED", "技能节点需要同时指定智能体与技能", node_id=node.id)
template = str(node.config.get("promptTemplate") or node.config.get("prompt_template") or "") 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]} 完成任务。" prompt = render_template(template, ctx.state()).strip() or f"请使用技能 {skill_names[0]} 完成任务。"
@@ -296,6 +301,7 @@ class SkillNodeExecutor(_AgentInvocation):
ctx, ctx,
agent_id=agent_id, agent_id=agent_id,
prompt=prompt, prompt=prompt,
model_name=model_name,
skill_names=skill_names, skill_names=skill_names,
response_mode="text", response_mode="text",
response_schema={}, response_schema={},
@@ -107,6 +107,7 @@ def test_workflow_http_api_and_coze_compat(monkeypatch: pytest.MonkeyPatch) -> N
class _Cfg: class _Cfg:
workflows = WorkflowConfig() workflows = WorkflowConfig()
models = []
monkeypatch.setattr(workflows, "get_app_config", lambda: _Cfg()) monkeypatch.setattr(workflows, "get_app_config", lambda: _Cfg())
monkeypatch.setattr(workflow_resources, "get_app_config", lambda: _Cfg()) monkeypatch.setattr(workflow_resources, "get_app_config", lambda: _Cfg())
@@ -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}) replay = await harness.client.post("/api/workflows/embed/verify", json={"ticket": ticket})
assert replay.status_code == 401 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"