feat(workflow): set_input + validate over node graph

tests/test_workflow.py

@@ -22,3 +22,35 @@ def test_load_template_returns_independent_copy():
     a["nodes"].append({"id": -999})
     b = workflow.load_template("t2v")
     assert {-999} & {n.get("id") for n in b["nodes"]} == set()
+
+
+def test_set_input_patches_widgets_values_in_place():
+    wf = workflow.load_template("t2v")
+    target_node = next(n for n in wf["nodes"] if n["type"] == "CLIPTextEncode")
+    workflow.set_input(wf, target_node["id"], 0, "new prompt text")
+    refetched = next(n for n in wf["nodes"] if n["id"] == target_node["id"])
+    assert refetched["widgets_values"][0] == "new prompt text"
+
+
+def test_set_input_raises_for_unknown_node():
+    wf = workflow.load_template("t2v")
+    with pytest.raises(KeyError, match="node id"):
+        workflow.set_input(wf, 999_999_999, 0, "x")
+
+
+def test_validate_accepts_canonical_template():
+    wf = workflow.load_template("t2v")
+    workflow.validate(wf)  # must not raise
+
+
+def test_validate_rejects_workflow_with_no_nodes():
+    wf = {"nodes": [], "links": []}
+    with pytest.raises(ValueError, match="no nodes"):
+        workflow.validate(wf)
+
+
+def test_validate_rejects_orphan_link():
+    wf = workflow.load_template("t2v")
+    wf["links"].append([99999, 1, 0, 999_999_999, 0, "INT"])  # destination doesn't exist
+    with pytest.raises(ValueError, match="orphan link"):
+        workflow.validate(wf)

workflow.py

@@ -17,3 +17,40 @@ def load_template(mode: str) -> dict[str, Any]:
         raise ValueError(f"unknown mode {mode!r}; expected one of {VALID_MODES}")
     path = WORKFLOWS_DIR / f"{mode}.json"
     return copy.deepcopy(json.loads(path.read_text()))
+
+
+def set_input(workflow: dict[str, Any], node_id: int, widget_index: int, value: Any) -> None:
+    """Patch a node's widgets_values in place.
+
+    Args:
+        workflow: A workflow dict (must have a "nodes" list).
+        node_id: The id of the node to patch.
+        widget_index: Position within the node's widgets_values list.
+        value: New value.
+
+    Raises:
+        KeyError: If no node with the given id exists.
+    """
+    for node in workflow["nodes"]:
+        if node.get("id") == node_id:
+            widgets = node.setdefault("widgets_values", [])
+            while len(widgets) <= widget_index:
+                widgets.append(None)
+            widgets[widget_index] = value
+            return
+    raise KeyError(f"node id {node_id} not found in workflow")
+
+
+def validate(workflow: dict[str, Any]) -> None:
+    """Static schema validation. Raises ValueError on the first problem found."""
+    nodes = workflow.get("nodes")
+    if not isinstance(nodes, list) or len(nodes) == 0:
+        raise ValueError("workflow has no nodes")
+
+    node_ids = {n.get("id") for n in nodes if "id" in n}
+    for link in workflow.get("links", []):
+        if not isinstance(link, list) or len(link) < 6:
+            raise ValueError(f"malformed link {link}")
+        _, src, _, dst, _, _ = link
+        if src not in node_ids or dst not in node_ids:
+            raise ValueError(f"orphan link {link}")