mirror of
https://github.com/langgenius/dify.git
synced 2026-10-04 13:47:55 +08:00
Charles YaoandClaude Opus 5
1b9d301faf
An LLM-supplied int node_id reached diff_graphs, which drops falsy and non-string ids from its node maps, so result.changes came back empty and build_change_set's fallback joined a list containing an int. The TypeError took the whole session to `failed` (ESQ1-271) after the mutation had already been committed to the draft. validate_intent_args now rejects a non-string or empty node id, so filter_applicable turns it into a rejection reason the agent can act on. build_change_set str()s defensively -- that join must never be what kills a session. Also extends the same guard to set_node_config's `path` arg: a non-string path used to reach _resolve_path's str.split() and raise AttributeError, surfacing "'int' object has no attribute 'split'" to the repair agent instead of an actionable rejection reason. Same boundary, same defect shape, deliberately included here rather than deferred. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
447 lines
19 KiB
Python
447 lines
19 KiB
Python
"""Pure, client-side graph mutation helpers.
|
|
|
|
Dify's ``sync_draft_workflow`` has no server-side patch primitive: it always
|
|
replaces the whole draft graph. The adapter (``dify_port.py``) therefore
|
|
reads the current graph, applies mutations like ``apply_set_node_config``
|
|
locally, and writes the whole mutated graph back. These helpers do the local
|
|
mutation only — no DB, no services, no I/O.
|
|
"""
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
from core.dify_builder.models import Graph, MutationIntent
|
|
|
|
MUTATION_ARG_KEYS: dict[str, tuple[str, ...]] = {
|
|
"set_node_config": ("node_id", "path", "value"),
|
|
"create_node": ("node_type", "config"),
|
|
"delete_node": ("node_id",),
|
|
"connect": ("from_node", "to_node"),
|
|
"insert_between": ("edge", "node_type", "config"),
|
|
}
|
|
|
|
# Args that name a node. An LLM occasionally emits these as ints or empty
|
|
# strings; unchecked they reach diff_graphs, which drops falsy/non-str ids
|
|
# from its node maps, and then crash build_change_set's join (ESQ1-271).
|
|
_NODE_ID_ARGS = ("node_id", "from_node", "to_node")
|
|
|
|
# `path` reaches _resolve_path's str.split; a non-string raises AttributeError
|
|
# there, so the rejection reason the repair agent reads is a Python internals
|
|
# message instead of an actionable one.
|
|
_STRING_ARGS = ("path",)
|
|
|
|
|
|
def validate_intent_args(intent: MutationIntent) -> None:
|
|
"""Raise ``ValueError`` if ``intent.args`` is missing a required key for
|
|
``intent.op``, or if ``intent.op`` isn't one of ``MUTATION_ARG_KEYS``'s
|
|
five recognized verbs. Optional keys (marked ``?`` on ``MutationIntent``)
|
|
are not checked here -- each ``apply_*`` function defaults them itself.
|
|
"""
|
|
required = MUTATION_ARG_KEYS.get(intent.op)
|
|
if required is None:
|
|
raise ValueError(f"unknown mutation op: {intent.op!r}")
|
|
missing = [key for key in required if key not in intent.args]
|
|
if missing:
|
|
raise ValueError(f"missing required arg(s) {missing} for op {intent.op!r}")
|
|
|
|
for key in _NODE_ID_ARGS:
|
|
if key not in intent.args:
|
|
continue
|
|
value = intent.args[key]
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"{key} must be a non-empty string for op {intent.op!r}, got {value!r}")
|
|
|
|
for key in _STRING_ARGS:
|
|
if key not in intent.args:
|
|
continue
|
|
value = intent.args[key]
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError(f"{key} must be a non-empty string for op {intent.op!r}, got {value!r}")
|
|
|
|
|
|
def _resolve_path(container: dict[str, Any], path: str) -> tuple[Any, str | int]:
|
|
"""Walk ``path``'s parent segments and return ``(parent, last_key)``.
|
|
|
|
Segments are dot-separated; an all-digit segment indexes a list, any
|
|
other segment is a mapping key. Only the FINAL segment may be absent --
|
|
that is the key being written. Every intermediate segment must already
|
|
resolve, and a segment that cannot be walked raises ``ValueError``.
|
|
|
|
Raising is the point. ``filter_applicable`` dry-runs each intent through
|
|
this function and turns the exception into a rejection reason the repair
|
|
agent sees on its next attempt. Before this existed, a dotted path wrote
|
|
one flat junk key that the workflow engine ignored, so every nested
|
|
repair silently did nothing (ESQ1-285, ESQ1-290).
|
|
"""
|
|
segments = path.split(".")
|
|
if not path or any(segment == "" for segment in segments):
|
|
raise ValueError(f"empty segment in path: {path!r}")
|
|
|
|
cursor: Any = container
|
|
for depth, segment in enumerate(segments[:-1]):
|
|
walked = ".".join(segments[: depth + 1])
|
|
if segment.isdigit():
|
|
if not isinstance(cursor, list):
|
|
raise ValueError(f"path {path!r}: {walked} indexes a {type(cursor).__name__}, not a list")
|
|
index = int(segment)
|
|
if index >= len(cursor):
|
|
raise ValueError(f"path {path!r}: index {index} out of range at {walked} (len {len(cursor)})")
|
|
cursor = cursor[index]
|
|
continue
|
|
if not isinstance(cursor, dict):
|
|
raise ValueError(f"path {path!r}: {walked} is not a mapping")
|
|
if segment not in cursor:
|
|
raise ValueError(f"path {path!r}: no key {segment!r} at {walked}")
|
|
cursor = cursor[segment]
|
|
|
|
last = segments[-1]
|
|
if last.isdigit():
|
|
if not isinstance(cursor, list):
|
|
raise ValueError(f"path {path!r}: final segment indexes a {type(cursor).__name__}, not a list")
|
|
index = int(last)
|
|
if index >= len(cursor):
|
|
raise ValueError(f"path {path!r}: index {index} out of range (len {len(cursor)})")
|
|
return cursor, index
|
|
if not isinstance(cursor, dict):
|
|
raise ValueError(f"path {path!r}: final container is a {type(cursor).__name__}, not a mapping")
|
|
return cursor, last
|
|
|
|
|
|
def apply_set_node_config(graph: Graph, node_id: str, path: str, value: Any) -> tuple[Graph, list[str]]:
|
|
"""Set the value at ``path`` inside ``node["data"]`` for the node whose
|
|
``id == node_id``. ``path`` is dot-separated (``"code"``,
|
|
``"cases.0.conditions.1.value"``); see ``_resolve_path``.
|
|
|
|
The placeholder agent emits intents shaped ``{node_id, path, value}``
|
|
(e.g. ``path="code"`` to rewrite a Code node's ``data["code"]``); this is
|
|
the mutation that consumes them.
|
|
|
|
Returns a ``(new_graph, changed_node_ids)`` tuple. ``graph`` is
|
|
deep-copied first and never mutated. Raises ``ValueError`` if no node in
|
|
``graph["nodes"]`` has a matching ``id``.
|
|
"""
|
|
new_graph = copy.deepcopy(graph)
|
|
|
|
for node in new_graph.get("nodes", []):
|
|
if node.get("id") == node_id:
|
|
parent, key = _resolve_path(node.setdefault("data", {}), path)
|
|
parent[key] = value
|
|
return new_graph, [node_id]
|
|
|
|
raise ValueError(f"node not found: {node_id}")
|
|
|
|
|
|
def _next_unique_id(base: str, existing_ids: set) -> str:
|
|
"""Return ``base`` if unused, else ``base_2``, ``base_3``, ... on
|
|
collision with ``existing_ids``. Shared bump sequence for node ids
|
|
(``_next_node_id``) and edge ids (``_make_edge``)."""
|
|
candidate = base
|
|
suffix = 1
|
|
while candidate in existing_ids:
|
|
suffix += 1
|
|
candidate = f"{base}_{suffix}"
|
|
return candidate
|
|
|
|
|
|
def _next_node_id(prefix: str, existing_ids: set) -> str:
|
|
"""Return a short, human-readable node id not present in ``existing_ids``.
|
|
|
|
Mirrors ``core.workflow.generator.runner._next_generated_node_id``'s
|
|
prefix, then prefix_2, prefix_3, ... collision scheme -- reimplemented
|
|
locally so this module stays free of any dependency on the generator
|
|
subsystem.
|
|
"""
|
|
return _next_unique_id(prefix, existing_ids)
|
|
|
|
|
|
def _default_position(graph: Graph) -> dict[str, float]:
|
|
"""Place a new node to the right of the rightmost existing node.
|
|
|
|
Slice 1 uses simple deterministic rightward placement, not the
|
|
generator's topology-aware layout (``_layout_top_level_nodes``) --
|
|
good enough for one node at a time; a smarter layout is Slice 2/3's
|
|
concern if the canvas ever needs it.
|
|
"""
|
|
nodes = graph.get("nodes", [])
|
|
if not nodes:
|
|
return {"x": 100.0, "y": 100.0}
|
|
max_x = max(float(n.get("position", {}).get("x", 0.0)) for n in nodes)
|
|
return {"x": max_x + 260.0, "y": 100.0}
|
|
|
|
|
|
def _build_node(
|
|
graph: Graph,
|
|
node_type: str,
|
|
config: dict[str, Any],
|
|
position: dict[str, float] | None,
|
|
node_id: str | None,
|
|
) -> dict[str, Any]:
|
|
"""Construct one ``GraphNodeDict``-shaped node, generating an id/position
|
|
when omitted (mirrors the generator's ``_fill_node_defaults``,
|
|
``core/workflow/generator/runner.py:2387-2394``). Raises ``ValueError``
|
|
if a caller-supplied ``node_id`` already exists in ``graph["nodes"]``.
|
|
"""
|
|
existing_ids = {n.get("id") for n in graph.get("nodes", [])}
|
|
if node_id is not None:
|
|
if node_id in existing_ids:
|
|
raise ValueError(f"node id already exists: {node_id}")
|
|
new_id = node_id
|
|
else:
|
|
new_id = _next_node_id(node_type, existing_ids)
|
|
|
|
data = copy.deepcopy(config)
|
|
data["type"] = node_type # data.type is the real node type -- never overridden by config
|
|
data.setdefault("title", new_id)
|
|
data.setdefault("desc", "")
|
|
data.setdefault("selected", False)
|
|
|
|
return {
|
|
"id": new_id,
|
|
"type": "custom",
|
|
"position": position if position is not None else _default_position(graph),
|
|
"data": data,
|
|
}
|
|
|
|
|
|
def apply_create_node(
|
|
graph: Graph,
|
|
node_type: str,
|
|
config: dict[str, Any],
|
|
position: dict[str, float] | None = None,
|
|
node_id: str | None = None,
|
|
) -> tuple[Graph, list[str]]:
|
|
"""Append a new ``GraphNodeDict``-shaped node to the graph.
|
|
|
|
``node_id`` is optional (not part of spec Sec 9's terse args list) --
|
|
when omitted a short id is generated from ``node_type``; when supplied,
|
|
a collision with an existing node id raises ``ValueError`` (the Slice 1
|
|
duplicate-id validation). Returns ``(new_graph, [new_node_id])``.
|
|
"""
|
|
new_graph = copy.deepcopy(graph)
|
|
node = _build_node(new_graph, node_type, config, position, node_id)
|
|
new_graph.setdefault("nodes", []).append(node)
|
|
return new_graph, [node["id"]]
|
|
|
|
|
|
def apply_delete_node(graph: Graph, node_id: str) -> tuple[Graph, list[str]]:
|
|
"""Remove the node with ``id == node_id`` and every edge touching it.
|
|
|
|
Raises ``ValueError`` if no node in ``graph["nodes"]`` has a matching
|
|
``id`` (mirrors ``apply_set_node_config``'s not-found behavior).
|
|
Returns ``(new_graph, [node_id])``.
|
|
"""
|
|
new_graph = copy.deepcopy(graph)
|
|
nodes = new_graph.get("nodes", [])
|
|
if not any(n.get("id") == node_id for n in nodes):
|
|
raise ValueError(f"node not found: {node_id}")
|
|
|
|
new_graph["nodes"] = [n for n in nodes if n.get("id") != node_id]
|
|
new_graph["edges"] = [
|
|
e for e in new_graph.get("edges", []) if e.get("source") != node_id and e.get("target") != node_id
|
|
]
|
|
return new_graph, [node_id]
|
|
|
|
|
|
def _make_edge(
|
|
source: str, target: str, source_handle: str | None, target_handle: str | None, existing_edge_ids: set
|
|
) -> dict:
|
|
"""Build one ``GraphEdgeDict``-shaped edge dict with a fresh
|
|
collision-free id (``source-target``, then ``source-target_2``, ... on
|
|
collision with ``existing_edge_ids``). Handles default to
|
|
"source"/"target" (mirrors the generator's ``_fill_edge_defaults``)."""
|
|
return {
|
|
"id": _next_unique_id(f"{source}-{target}", existing_edge_ids),
|
|
"source": source,
|
|
"target": target,
|
|
"type": "custom",
|
|
"sourceHandle": source_handle or "source",
|
|
"targetHandle": target_handle or "target",
|
|
}
|
|
|
|
|
|
def apply_connect(
|
|
graph: Graph,
|
|
from_node: str,
|
|
to_node: str,
|
|
source_handle: str | None = None,
|
|
target_handle: str | None = None,
|
|
) -> tuple[Graph, list[str]]:
|
|
"""Add a ``GraphEdgeDict``-shaped edge between two existing nodes.
|
|
|
|
Raises ``ValueError`` if either ``from_node`` or ``to_node`` is not a
|
|
node id present in ``graph["nodes"]`` (the Slice 1 dangling-ref
|
|
validation). Handles default to "source"/"target" (mirrors the
|
|
generator's ``_fill_edge_defaults``). Returns
|
|
``(new_graph, [from_node, to_node])``.
|
|
"""
|
|
new_graph = copy.deepcopy(graph)
|
|
node_ids = {n.get("id") for n in new_graph.get("nodes", [])}
|
|
if from_node not in node_ids:
|
|
raise ValueError(f"node not found: {from_node}")
|
|
if to_node not in node_ids:
|
|
raise ValueError(f"node not found: {to_node}")
|
|
|
|
existing_edge_ids = {e.get("id") for e in new_graph.get("edges", [])}
|
|
edge = _make_edge(from_node, to_node, source_handle, target_handle, existing_edge_ids)
|
|
new_graph.setdefault("edges", []).append(edge)
|
|
return new_graph, [from_node, to_node]
|
|
|
|
|
|
def apply_insert_between(
|
|
graph: Graph,
|
|
edge: dict[str, str],
|
|
node_type: str,
|
|
config: dict[str, Any],
|
|
position: dict[str, float] | None = None,
|
|
node_id: str | None = None,
|
|
) -> tuple[Graph, list[str]]:
|
|
"""Split an existing edge with a new node: remove the old edge, add
|
|
``old_source -> new_node`` and ``new_node -> old_target``.
|
|
|
|
``edge`` identifies the edge to split by ``{"source": ..., "target": ...}``
|
|
(matching Dify's own edge fields, not a caller-known internal edge id).
|
|
Raises ``ValueError`` if no edge in ``graph["edges"]`` matches (the
|
|
Slice 1 dangling-ref validation). Returns ``(new_graph, [new_node_id])``.
|
|
"""
|
|
old_source = edge.get("source")
|
|
old_target = edge.get("target")
|
|
new_graph = copy.deepcopy(graph)
|
|
edges = new_graph.get("edges", [])
|
|
matched = next((e for e in edges if e.get("source") == old_source and e.get("target") == old_target), None)
|
|
if matched is None:
|
|
raise ValueError(f"edge not found: {old_source} -> {old_target}")
|
|
|
|
node = _build_node(new_graph, node_type, config, position, node_id)
|
|
new_id = node["id"]
|
|
new_graph.setdefault("nodes", []).append(node)
|
|
|
|
remaining_edges = [e for e in edges if e is not matched]
|
|
existing_edge_ids = {e.get("id") for e in remaining_edges}
|
|
incoming = _make_edge(old_source, new_id, matched.get("sourceHandle"), "target", existing_edge_ids)
|
|
existing_edge_ids.add(incoming["id"])
|
|
outgoing = _make_edge(new_id, old_target, "source", matched.get("targetHandle"), existing_edge_ids)
|
|
new_graph["edges"] = [*remaining_edges, incoming, outgoing]
|
|
return new_graph, [new_id]
|
|
|
|
|
|
def diff_graphs(before: Graph, after: Graph) -> tuple[list[str], str]:
|
|
"""Compare two graphs and return ``(changes, scope)``.
|
|
|
|
``scope`` is ``"structure"`` if any node or edge was added or removed,
|
|
else ``"configuration"`` if any surviving node's ``data`` changed, else
|
|
``"configuration"`` with an empty ``changes`` list if nothing differs.
|
|
``changes`` is human-readable lines (e.g. ``"added node knowledge-1"``,
|
|
``"added knowledge-1 → llm-1"``, ``"llm-1: prompt_template updated"``)
|
|
-- not bare ids.
|
|
"""
|
|
before_nodes = {n.get("id"): n for n in before.get("nodes", []) if n.get("id")}
|
|
after_nodes = {n.get("id"): n for n in after.get("nodes", []) if n.get("id")}
|
|
|
|
def _edge_key(e: dict[str, Any]) -> tuple[str, str, str, str]:
|
|
return (
|
|
str(e.get("source")),
|
|
str(e.get("target")),
|
|
str(e.get("sourceHandle") or "source"),
|
|
str(e.get("targetHandle") or "target"),
|
|
)
|
|
|
|
before_edges = {_edge_key(e) for e in before.get("edges", [])}
|
|
after_edges = {_edge_key(e) for e in after.get("edges", [])}
|
|
|
|
added_nodes = sorted(after_nodes.keys() - before_nodes.keys())
|
|
removed_nodes = sorted(before_nodes.keys() - after_nodes.keys())
|
|
added_edges = sorted(after_edges - before_edges)
|
|
removed_edges = sorted(before_edges - after_edges)
|
|
|
|
changes: list[str] = []
|
|
changes.extend(f"added node {node_id}" for node_id in added_nodes)
|
|
changes.extend(f"removed node {node_id}" for node_id in removed_nodes)
|
|
changes.extend(f"added {source} → {target}" for source, target, _sh, _th in added_edges)
|
|
changes.extend(f"removed {source} → {target}" for source, target, _sh, _th in removed_edges)
|
|
|
|
for node_id in sorted(before_nodes.keys() & after_nodes.keys()):
|
|
before_data = before_nodes[node_id].get("data", {})
|
|
after_data = after_nodes[node_id].get("data", {})
|
|
changed_keys = sorted(
|
|
key for key in set(before_data) | set(after_data) if before_data.get(key) != after_data.get(key)
|
|
)
|
|
changes.extend(f"{node_id}: {key} updated" for key in changed_keys)
|
|
|
|
structural = bool(added_nodes or removed_nodes or added_edges or removed_edges)
|
|
return changes, "structure" if structural else "configuration"
|
|
|
|
|
|
def structural_fingerprint(graph: Graph) -> str:
|
|
"""Stable hash of a graph's STRUCTURE only -- node identity+type and edge
|
|
topology -- EXCLUDING node ``data`` config. Two graphs with the same
|
|
nodes/edges but different node configs share a fingerprint; adding,
|
|
removing, or reconnecting a node changes it. Used by recovery (C-1) to
|
|
tell a config-only hand-edit from a structural one. Order-independent
|
|
(nodes and edges are sorted before hashing)."""
|
|
nodes = sorted((str(n.get("id", "")), str((n.get("data") or {}).get("type", ""))) for n in graph.get("nodes", []))
|
|
edges = sorted(
|
|
(
|
|
str(e.get("source", "")),
|
|
str(e.get("target", "")),
|
|
str(e.get("sourceHandle", "")),
|
|
str(e.get("targetHandle", "")),
|
|
)
|
|
for e in graph.get("edges", [])
|
|
)
|
|
canonical = json.dumps({"nodes": nodes, "edges": edges}, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def node_ids(graph: Graph) -> list[str]:
|
|
"""The ids of every node in ``graph``, in document order (missing-id
|
|
nodes skipped). Used by recovery's target-presence check."""
|
|
return [str(n["id"]) for n in graph.get("nodes", []) if n.get("id") is not None]
|
|
|
|
|
|
STRUCTURAL_OPS: frozenset[str] = frozenset({"create_node", "delete_node", "connect", "insert_between"})
|
|
|
|
APPLY_FNS: dict[str, Any] = {
|
|
"set_node_config": apply_set_node_config,
|
|
"create_node": apply_create_node,
|
|
"delete_node": apply_delete_node,
|
|
"connect": apply_connect,
|
|
"insert_between": apply_insert_between,
|
|
}
|
|
|
|
|
|
def filter_applicable(
|
|
graph: Graph,
|
|
intents: list[MutationIntent],
|
|
allowed_node_types: set[str] | None = None,
|
|
) -> tuple[list[MutationIntent], list[tuple[MutationIntent, str]]]:
|
|
"""Dry-run each intent through the real validate_intent_args + apply_* on a
|
|
working deep copy, in order (so a connect sees a node an earlier create_node
|
|
added). Returns (applicable, rejected) where each rejected entry is
|
|
(intent, reason).
|
|
|
|
Catches every failure the live apply_repair would hit -- unknown op, missing
|
|
required arg / extra arg key, dangling node/edge ref, duplicate id -- and
|
|
additionally rejects a create_node / insert_between whose node_type is not in
|
|
allowed_node_types (a check apply_* does not do). Applicable intents advance
|
|
the working copy; a rejected intent does not.
|
|
"""
|
|
working = copy.deepcopy(graph)
|
|
applicable: list[MutationIntent] = []
|
|
rejected: list[tuple[MutationIntent, str]] = []
|
|
for intent in intents:
|
|
try:
|
|
validate_intent_args(intent)
|
|
if allowed_node_types is not None and intent.op in ("create_node", "insert_between"):
|
|
node_type = intent.args.get("node_type")
|
|
if node_type not in allowed_node_types:
|
|
rejected.append((intent, f"node_type not allowed: {node_type!r}"))
|
|
continue
|
|
working, _changed = APPLY_FNS[intent.op](working, **intent.args)
|
|
except Exception as exc:
|
|
rejected.append((intent, str(exc)))
|
|
continue
|
|
applicable.append(intent)
|
|
return applicable, rejected
|