代码提交

offline-backend-20260512/backend/tests/test_workflow_engine.py

@@ -0,0 +1,688 @@
+"""Phase 2/3 tests: expressions, scheduler semantics, policies."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from deerflow.config.workflow_config import WorkflowConfig
+from deerflow.persistence.workflow_events import MemoryWorkflowEventStore
+from deerflow.persistence.workflow_runs import MemoryWorkflowRunStore
+from deerflow.workflows.errors import WorkflowError
+from deerflow.workflows.expressions import evaluate_condition, render_template, render_value
+from deerflow.workflows.nodes import build_default_registry
+from deerflow.workflows.runtime import (
+    CancelToken,
+    PersistingWorkflowEventSink,
+    RunContext,
+    WorkflowEngine,
+    WorkflowPaused,
+    WorkflowRuntimeDeps,
+)
+from deerflow.workflows.schemas import WorkflowGraph
+from deerflow.workflows.security.http_policy import enforce_http_target
+from deerflow.workflows.security.sql_policy import assert_read_only, assert_tables_allowed, enforce_limit
+from deerflow.workflows.validator import validate_workflow_graph
+
+
+def _graph(nodes: list[dict[str, Any]], edges: list[dict[str, Any]], **kwargs: Any) -> WorkflowGraph:
+    return WorkflowGraph.model_validate(
+        {
+            "schemaVersion": "1.0",
+            "id": "wf-test",
+            "name": "test",
+            "nodes": nodes,
+            "edges": edges,
+            **kwargs,
+        }
+    )
+
+
+async def _run(
+    graph: WorkflowGraph,
+    inputs: dict[str, Any] | None = None,
+    *,
+    deps: WorkflowRuntimeDeps | None = None,
+    resume_payload: dict[str, Any] | None = None,
+    config: WorkflowConfig | None = None,
+) -> tuple[Any, list[dict[str, Any]]]:
+    run_store = MemoryWorkflowRunStore()
+    event_store = MemoryWorkflowEventStore()
+    published: list[dict[str, Any]] = []
+
+    async def publish(event: dict[str, Any]) -> None:
+        published.append(event)
+
+    sink = PersistingWorkflowEventSink(event_store, run_id="run-1", workflow_id=graph.id, version_id="v1", publish=publish)
+    ctx = RunContext(
+        run_id="run-1",
+        workflow_id=graph.id,
+        version_id="v1",
+        owner_id="user-a",
+        graph=graph,
+        inputs=inputs or {},
+        deps=deps or WorkflowRuntimeDeps(),
+        cancel=CancelToken(),
+        emit=sink.emit,
+        resume_payload=resume_payload,
+    )
+    engine = WorkflowEngine(build_default_registry(config or WorkflowConfig()), run_store=run_store)
+    result = await engine.run(ctx)
+    return result, published
+
+
+# ── expressions ─────────────────────────────────────────────────────────
+
+
+def test_expression_rendering_and_paths() -> None:
+    state = {
+        "inputs": {"topic": "储能"},
+        "nodes": {"n1": {"data": {"items": [{"title": "A"}, {"title": "B"}], "score": 0.9}}},
+        "run": {"id": "r1"},
+        "loop": {},
+        "env": {},
+    }
+    assert render_template("主题：{{ inputs.topic }}", state) == "主题：储能"
+    assert render_template("{{ nodes.n1.data.items[1].title }}", state) == "B"
+    # A lone placeholder keeps the native type.
+    assert render_value("{{ nodes.n1.data.score }}", state) == 0.9
+    assert render_value({"a": ["{{ inputs.topic }}"]}, state) == {"a": ["储能"]}
+    # Missing paths resolve to empty rather than exploding.
+    assert render_template("{{ nodes.nope.data.x }}", state) == ""
+
+
+def test_expression_rejects_unknown_roots_and_code() -> None:
+    state = {"inputs": {}, "nodes": {}, "run": {}, "loop": {}, "env": {}}
+    with pytest.raises(WorkflowError) as exc:
+        render_template("{{ __import__('os').system('id') }}", state)
+    assert exc.value.code == "WORKFLOW_EXPRESSION_INVALID"
+    with pytest.raises(WorkflowError):
+        render_template("{{ os.environ.SECRET }}", state)
+
+
+def test_condition_evaluation() -> None:
+    state = {
+        "inputs": {"count": 5, "tags": ["a", "b"], "name": "储能报告"},
+        "nodes": {"n1": {"data": {"ok": True}}},
+        "run": {},
+        "loop": {"n1": 2},
+        "env": {},
+    }
+    assert evaluate_condition("inputs.count > 3", state) is True
+    assert evaluate_condition("inputs.count > 3 and nodes.n1.data.ok", state) is True
+    assert evaluate_condition("inputs.count > 9 or nodes.n1.data.ok == true", state) is True
+    assert evaluate_condition("'a' in inputs.tags", state) is True
+    assert evaluate_condition("inputs.name contains '储能'", state) is True
+    assert evaluate_condition("not nodes.n1.data.ok", state) is False
+    assert evaluate_condition("loop.n1 < 5", state) is True
+
+
+# ── scheduler ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_start_output_skeleton_emits_ordered_events() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start", "name": "开始"},
+            {"id": "end", "type": "output", "name": "输出", "config": {"mapping": {"echo": "{{ inputs.topic }}"}}},
+        ],
+        [{"id": "e1", "source": "start", "target": "end"}],
+    )
+    result, events = await _run(graph, {"topic": "储能"})
+    assert result.output == {"echo": "储能"}
+    seqs = [e["seq"] for e in events]
+    assert seqs == sorted(seqs) and seqs[0] == 1
+    kinds = [e["event"] for e in events]
+    assert kinds == [
+        "node.started",
+        "node.completed",
+        "node.started",
+        "node.completed",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_condition_prunes_untaken_branch() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "gate",
+                "type": "condition",
+                "config": {
+                    "branches": [{"name": "big", "expression": "inputs.n > 10"}],
+                    "defaultBranch": "small",
+                },
+            },
+            {"id": "big", "type": "transform", "config": {"operations": [{"op": "set", "target": "path", "value": "big"}]}},
+            {"id": "small", "type": "transform", "config": {"operations": [{"op": "set", "target": "path", "value": "small"}]}},
+            {"id": "end", "type": "output", "config": {"mapping": {"path": "{{ nodes.small.data.path }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "gate"},
+            {"id": "e2", "source": "gate", "target": "big", "sourcePort": "big"},
+            {"id": "e3", "source": "gate", "target": "small", "sourcePort": "small"},
+            {"id": "e4", "source": "small", "target": "end"},
+        ],
+    )
+    result, _ = await _run(graph, {"n": 1})
+    assert result.output == {"path": "small"}
+    assert "big" not in result.node_results
+
+
+@pytest.mark.asyncio
+async def test_parallel_fanout_joins_at_merge() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {"id": "a", "type": "transform", "config": {"operations": [{"op": "set", "target": "v", "value": "A"}]}},
+            {"id": "b", "type": "transform", "config": {"operations": [{"op": "set", "target": "v", "value": "B"}]}},
+            {"id": "join", "type": "merge", "config": {"strategy": "all"}},
+            {"id": "end", "type": "output", "config": {"mapping": {"joined": "{{ nodes.join.data.succeeded }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "a"},
+            {"id": "e2", "source": "start", "target": "b"},
+            {"id": "e3", "source": "a", "target": "join"},
+            {"id": "e4", "source": "b", "target": "join"},
+            {"id": "e5", "source": "join", "target": "end"},
+        ],
+    )
+    result, _ = await _run(graph)
+    assert sorted(result.output["joined"]) == ["a", "b"]
+
+
+@pytest.mark.asyncio
+async def test_loop_reruns_body_and_respects_cap() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {"id": "body", "type": "transform", "config": {"operations": [{"op": "set", "target": "tick", "value": "1"}]}},
+            {
+                "id": "gate",
+                "type": "loop",
+                "config": {"maxIterations": 3, "continueWhen": "true", "bodyEntry": "body"},
+            },
+            {"id": "end", "type": "output", "config": {"mapping": {"iterations": "{{ nodes.gate.data.iteration }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "body"},
+            {"id": "e2", "source": "body", "target": "gate"},
+            {"id": "e3", "source": "gate", "target": "body", "sourcePort": "continue"},
+            {"id": "e4", "source": "gate", "target": "end", "sourcePort": "done"},
+        ],
+        settings={"maxLoopIterations": 5},
+    )
+    result, _ = await _run(graph)
+    # maxIterations=3 means the gate stops handing out the continue port at 2.
+    assert result.output["iterations"] == 2
+
+
+@pytest.mark.asyncio
+async def test_max_steps_limit_is_enforced() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {"id": "a", "type": "transform"},
+            {"id": "end", "type": "output"},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "a"},
+            {"id": "e2", "source": "a", "target": "end"},
+        ],
+        settings={"maxSteps": 2},
+    )
+    with pytest.raises(WorkflowError) as exc:
+        await _run(graph)
+    assert exc.value.code == "WORKFLOW_MAX_STEPS"
+
+
+@pytest.mark.asyncio
+async def test_human_input_pauses_then_resumes() -> None:
+    nodes = [
+        {"id": "start", "type": "start"},
+        {
+            "id": "ask",
+            "type": "human_input",
+            "config": {"prompt": "请确认 {{ inputs.topic }}", "formSchema": {"required": ["comment"]}, "actions": ["submit"]},
+        },
+        {"id": "end", "type": "output", "config": {"mapping": {"comment": "{{ nodes.ask.data.values.comment }}"}}},
+    ]
+    edges = [
+        {"id": "e1", "source": "start", "target": "ask"},
+        {"id": "e2", "source": "ask", "target": "end"},
+    ]
+    graph = _graph(nodes, edges)
+    with pytest.raises(WorkflowPaused) as paused:
+        await _run(graph, {"topic": "储能"})
+    assert paused.value.node_id == "ask"
+    assert paused.value.pending_input["prompt"] == "请确认 储能"
+    assert paused.value.pending_input["formSchema"]["required"] == ["comment"]
+
+    result, _ = await _run(
+        graph,
+        {"topic": "储能"},
+        resume_payload={"nodeId": "ask", "action": "submit", "values": {"comment": "同意"}},
+    )
+    assert result.output == {"comment": "同意"}
+
+
+@pytest.mark.asyncio
+async def test_human_input_rejects_missing_required_field() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {"id": "ask", "type": "human_input", "config": {"formSchema": {"required": ["comment"]}}},
+            {"id": "end", "type": "output"},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "ask"},
+            {"id": "e2", "source": "ask", "target": "end"},
+        ],
+    )
+    with pytest.raises(WorkflowError) as exc:
+        await _run(graph, resume_payload={"nodeId": "ask", "action": "submit", "values": {}})
+    assert exc.value.code == "WORKFLOW_HUMAN_INPUT_INVALID"
+
+
+@pytest.mark.asyncio
+async def test_node_failure_can_be_tolerated_with_on_error_continue() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "boom",
+                "type": "transform",
+                "config": {"onError": "continue", "operations": [{"op": "not_a_real_op"}]},
+            },
+            {"id": "end", "type": "output", "config": {"mapping": {"ok": "true"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "boom"},
+            {"id": "e2", "source": "boom", "target": "end"},
+        ],
+    )
+    result, events = await _run(graph)
+    assert result.output == {"ok": "true"}
+    assert result.node_results["boom"]["metadata"]["failed"] is True
+    assert any(e["event"] == "node.failed" for e in events)
+
+
+@pytest.mark.asyncio
+async def test_agent_node_streams_deltas_and_parses_json() -> None:
+    async def run_agent(**kwargs: Any) -> dict[str, Any]:
+        on_delta = kwargs["on_delta"]
+        await on_delta("思考")
+        await on_delta("完成")
+        assert "储能" in kwargs["prompt"]
+        return {"text": '```json\n{"title": "储能报告"}\n```', "usage": {"tokens": 12}}
+
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "writer",
+                "type": "agent",
+                "config": {
+                    "agentId": "a1",
+                    "promptTemplate": "写一份关于 {{ inputs.topic }} 的报告",
+                    "responseMode": "json",
+                    "responseSchema": {"required": ["title"]},
+                    "parallelGroupId": "round-1",
+                    "roundIndex": 1,
+                    "agentName": "报告撰写智能体",
+                    "agentAvatarUrl": "https://assets.example.com/writer.png",
+                },
+            },
+            {"id": "end", "type": "output", "config": {"mapping": {"title": "{{ nodes.writer.data.json.title }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "writer"},
+            {"id": "e2", "source": "writer", "target": "end"},
+        ],
+    )
+    result, events = await _run(graph, {"topic": "储能"}, deps=WorkflowRuntimeDeps(run_agent=run_agent))
+    assert result.output == {"title": "储能报告"}
+    deltas = [(e["data"].get("delta") or e["data"].get("text")) for e in events if e["event"] == "node.output.delta"]
+    assert "".join(deltas) == "思考完成"
+    assert all(e["data"].get("channel") == "answer" for e in events if e["event"] == "node.output.delta")
+    started = next(event for event in events if event["event"] == "node.started" and event["nodeId"] == "writer")
+    assert started["data"]["nodeKind"] == "agent"
+    assert started["data"]["parallelGroupId"] == "round-1"
+    assert started["data"]["roundIndex"] == 1
+    assert started["data"]["agent"] == {
+        "agentId": "a1",
+        "name": "报告撰写智能体",
+        "avatarUrl": "https://assets.example.com/writer.png",
+    }
+
+
+@pytest.mark.asyncio
+async def test_missing_capability_becomes_typed_error() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {"id": "writer", "type": "agent", "config": {"agentId": "a1", "promptTemplate": "hi"}},
+        ],
+        [{"id": "e1", "source": "start", "target": "writer"}],
+    )
+    with pytest.raises(WorkflowError) as exc:
+        await _run(graph)
+    assert exc.value.code == "WORKFLOW_RESOURCE_MISSING"
+
+
+@pytest.mark.asyncio
+async def test_cancellation_stops_the_run() -> None:
+    run_store = MemoryWorkflowRunStore()
+    event_store = MemoryWorkflowEventStore()
+    graph = _graph(
+        [{"id": "start", "type": "start"}, {"id": "end", "type": "output"}],
+        [{"id": "e1", "source": "start", "target": "end"}],
+    )
+    sink = PersistingWorkflowEventSink(event_store, run_id="run-x", workflow_id=graph.id, version_id="v1")
+    cancel = CancelToken()
+    cancel.cancel()
+    ctx = RunContext(
+        run_id="run-x",
+        workflow_id=graph.id,
+        version_id="v1",
+        owner_id="u",
+        graph=graph,
+        inputs={},
+        deps=WorkflowRuntimeDeps(),
+        cancel=cancel,
+        emit=sink.emit,
+    )
+    engine = WorkflowEngine(build_default_registry(), run_store=run_store)
+    with pytest.raises(WorkflowError) as exc:
+        await engine.run(ctx)
+    assert exc.value.code == "WORKFLOW_CANCELLED"
+
+
+# ── policies ────────────────────────────────────────────────────────────
+
+
+def test_sql_policy_blocks_writes_and_stacked_statements() -> None:
+    assert assert_read_only("SELECT id FROM t WHERE x = :x").startswith("SELECT")
+    assert assert_read_only("WITH c AS (SELECT 1) SELECT * FROM c").startswith("WITH")
+    for bad in (
+        "DELETE FROM t",
+        "SELECT 1; DROP TABLE t",
+        "SELECT 1 -- ;\n; DROP TABLE t",
+        "SELECT * FROM t INTO OUTFILE '/tmp/x'",
+        "SELECT pg_sleep(10)",
+        "UPDATE t SET a = 1",
+    ):
+        with pytest.raises(WorkflowError) as exc:
+            assert_read_only(bad)
+        assert exc.value.code == "WORKFLOW_SQL_POLICY_DENIED"
+
+
+def test_sql_limit_is_appended_once() -> None:
+    assert enforce_limit("SELECT 1", 10) == "SELECT 1 LIMIT 10"
+    assert enforce_limit("SELECT 1 LIMIT 5", 10) == "SELECT 1 LIMIT 5"
+
+
+def test_sql_table_allowlist_blocks_unlisted_tables() -> None:
+    assert_tables_allowed("SELECT id FROM report_source", ["report_source"])
+    with pytest.raises(WorkflowError) as exc:
+        assert_tables_allowed("SELECT id FROM secrets", ["report_source"])
+    assert exc.value.code == "WORKFLOW_SQL_POLICY_DENIED"
+
+
+def test_downstream_node_reference_is_rejected_at_publish() -> None:
+    issues = validate_workflow_graph(
+        {
+            "schemaVersion": "1.0",
+            "id": "wf_ref",
+            "nodes": [
+                {"id": "start", "type": "start"},
+                {
+                    "id": "early",
+                    "type": "transform",
+                    "config": {"operations": [{"op": "set", "target": "x", "value": "{{ nodes.late.data.x }}"}]},
+                },
+                {
+                    "id": "late",
+                    "type": "transform",
+                    "config": {"operations": [{"op": "set", "target": "x", "value": "1"}]},
+                },
+                {"id": "end", "type": "output", "config": {"mapping": {"x": "{{ nodes.late.data.x }}"}}},
+            ],
+            "edges": [
+                {"id": "e1", "source": "start", "target": "early"},
+                {"id": "e2", "source": "early", "target": "late"},
+                {"id": "e3", "source": "late", "target": "end"},
+            ],
+        }
+    )
+    assert any(i.code == "WORKFLOW_EXPRESSION_INVALID" and i.node_id == "early" for i in issues)
+
+
+@pytest.mark.asyncio
+async def test_http_policy_blocks_private_and_metadata_targets() -> None:
+    policy = WorkflowConfig().http
+    for url in (
+        "http://127.0.0.1:8001/admin",
+        "http://169.254.169.254/latest/meta-data/",
+        "http://localhost/",
+        "file:///etc/passwd",
+        "http://example.com:22/",
+    ):
+        with pytest.raises(WorkflowError) as exc:
+            await enforce_http_target(url, policy)
+        assert exc.value.code in ("WORKFLOW_HTTP_POLICY_DENIED", "WORKFLOW_HTTP_FAILED")
+
+
+@pytest.mark.asyncio
+async def test_http_policy_allows_private_when_configured() -> None:
+    policy = WorkflowConfig().http.model_copy(update={"allow_private_networks": True, "allowed_ports": [8001]})
+    target = await enforce_http_target("http://127.0.0.1:8001/ping", policy)
+    assert target["host"] == "127.0.0.1"
+
+
+@pytest.mark.asyncio
+async def test_event_sink_persists_before_publishing() -> None:
+    store = MemoryWorkflowEventStore()
+    seen: list[int] = []
+
+    async def publish(event: dict[str, Any]) -> None:
+        # Persisted rows must already be readable when the live copy arrives.
+        rows = await store.list_after("r1", after_seq=0)
+        seen.append(len(rows))
+
+    sink = PersistingWorkflowEventSink(store, run_id="r1", workflow_id="w1", version_id="v1", publish=publish)
+    await sink.emit("run.started")
+    await sink.emit("run.completed")
+    assert seen == [1, 2]
+    assert [r["seq"] for r in await store.list_after("r1")] == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_run_store_lease_and_idempotency() -> None:
+    from datetime import UTC, datetime, timedelta
+
+    store = MemoryWorkflowRunStore()
+    payload = {
+        "workflow_id": "w1",
+        "workflow_version_id": "v1",
+        "owner_id": "u1",
+        "idempotency_key": "k1",
+        "input": {"a": 1},
+    }
+    run, created = await store.create_run(payload)
+    assert created is True
+    again, created_again = await store.create_run(payload)
+    assert created_again is False and again["id"] == run["id"]
+
+    until = datetime.now(UTC) + timedelta(seconds=30)
+    claimed = await store.claim_run(run["id"], lease_owner="w-a", lease_until=until)
+    assert claimed is not None and claimed["status"] == "running"
+    # A second worker cannot steal a live lease.
+    assert await store.claim_run(run["id"], lease_owner="w-b", lease_until=until) is None
+    assert await store.renew_lease(run["id"], lease_owner="w-a", lease_until=until) is True
+    assert await store.renew_lease(run["id"], lease_owner="w-b", lease_until=until) is False
+
+    # Expired lease is reclaimable (crash recovery).
+    await store.renew_lease(run["id"], lease_owner="w-a", lease_until=datetime.now(UTC) - timedelta(seconds=1))
+    assert await store.claim_run(run["id"], lease_owner="w-b", lease_until=until) is not None
+
+    paused = await store.set_awaiting_input(run["id"], lease_owner="w-b", pending_input={"nodeId": "ask"}, resume_token="t1")
+    assert paused is not None and paused["status"] == "awaiting_input"
+    assert await store.consume_resume_token(run["id"], "wrong") is None
+    resumed = await store.consume_resume_token(run["id"], "t1")
+    assert resumed is not None and resumed["status"] == "queued"
+    # One-shot: the same token cannot be replayed.
+    assert await store.consume_resume_token(run["id"], "t1") is None
+
+
+@pytest.mark.asyncio
+async def test_code_node_disabled_by_default() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {"id": "calc", "type": "code", "config": {"source": "output['x'] = 1"}},
+        ],
+        [{"id": "e1", "source": "start", "target": "calc"}],
+    )
+    with pytest.raises(WorkflowError) as exc:
+        await _run(graph)
+    assert exc.value.code == "WORKFLOW_CODE_SANDBOX_FAILED"
+
+
+@pytest.mark.asyncio
+async def test_code_node_executes_when_enabled() -> None:
+    config = WorkflowConfig()
+    config.code.enabled = True
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "calc",
+                "type": "code",
+                "config": {
+                    "source": "output['doubled'] = inputs['n'] * 2",
+                    "inputs": {"n": "{{ inputs.n }}"},
+                },
+            },
+            {"id": "end", "type": "output", "config": {"mapping": {"doubled": "{{ nodes.calc.data.doubled }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "calc"},
+            {"id": "e2", "source": "calc", "target": "end"},
+        ],
+    )
+    result, _ = await _run(graph, {"n": 21}, config=config)
+    assert result.output == {"doubled": 42}
+
+
+@pytest.mark.asyncio
+async def test_code_node_cannot_open_sockets() -> None:
+    config = WorkflowConfig()
+    config.code.enabled = True
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "calc",
+                "type": "code",
+                "config": {"source": "import socket\nsocket.socket()"},
+            },
+        ],
+        [{"id": "e1", "source": "start", "target": "calc"}],
+    )
+    with pytest.raises(WorkflowError) as exc:
+        await _run(graph, config=config)
+    assert exc.value.code == "WORKFLOW_CODE_SANDBOX_FAILED"
+
+
+def test_asyncio_smoke() -> None:
+    # Guards against an event-loop policy regression on Windows runners.
+    assert asyncio.get_event_loop_policy() is not None
+
+
+@pytest.mark.asyncio
+async def test_skill_node_invokes_agent_with_restricted_skills() -> None:
+    seen: dict[str, Any] = {}
+
+    async def run_agent(**kwargs: Any) -> dict[str, Any]:
+        seen.update(kwargs)
+        return {"text": "已完成检索", "thread_id": "thr-1"}
+
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "skill",
+                "type": "skill",
+                "config": {
+                    "agentId": "collector",
+                    "skillNames": ["web_search"],
+                    "promptTemplate": "查 {{ inputs.topic }}",
+                    "mode": "agent_skill",
+                },
+            },
+            {"id": "end", "type": "output", "config": {"mapping": {"text": "{{ nodes.skill.data.text }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "skill"},
+            {"id": "e2", "source": "skill", "target": "end"},
+        ],
+    )
+    result, _published = await _run(graph, {"topic": "储能"}, deps=WorkflowRuntimeDeps(run_agent=run_agent))
+    assert result.output == {"text": "已完成检索"}
+    assert seen["agent_id"] == "collector"
+    assert seen["skill_names"] == ["web_search"]
+    assert "储能" in seen["prompt"]
+
+
+@pytest.mark.asyncio
+async def test_callable_skill_mode_is_rejected() -> None:
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "skill",
+                "type": "skill",
+                "config": {"agentId": "collector", "skillNames": ["web_search"], "mode": "callable_skill"},
+            },
+        ],
+        [{"id": "e1", "source": "start", "target": "skill"}],
+    )
+    with pytest.raises(WorkflowError) as exc:
+        await _run(graph)
+    assert exc.value.code == "WORKFLOW_SKILL_FAILED"
+
+
+@pytest.mark.asyncio
+async def test_subworkflow_node_folds_child_output() -> None:
+    async def run_subworkflow(**kwargs: Any) -> dict[str, Any]:
+        assert kwargs["workflow_id"] == "child-wf"
+        assert kwargs["version_id"] == "child-v1"
+        assert kwargs["inputs"] == {"topic": "储能"}
+        return {"run_id": "child-run", "status": "completed", "output": {"title": "子报告"}}
+
+    graph = _graph(
+        [
+            {"id": "start", "type": "start"},
+            {
+                "id": "child",
+                "type": "subworkflow",
+                "config": {
+                    "workflowId": "child-wf",
+                    "versionId": "child-v1",
+                    "inputMapping": {"topic": "{{ inputs.topic }}"},
+                },
+            },
+            {"id": "end", "type": "output", "config": {"mapping": {"title": "{{ nodes.child.data.title }}"}}},
+        ],
+        [
+            {"id": "e1", "source": "start", "target": "child"},
+            {"id": "e2", "source": "child", "target": "end"},
+        ],
+    )
+    result, _ = await _run(graph, {"topic": "储能"}, deps=WorkflowRuntimeDeps(run_subworkflow=run_subworkflow))
+    assert result.output == {"title": "子报告"}