avoid serializing field: null + docs

src/wf_core/models/steps.py

@@ -14,8 +14,29 @@ class NodeUse(BaseModel):
     type: Literal["node"]
     node: str
     desc: str | None = None
-    in_map: dict[str, str] = Field(default_factory=dict)
-    out_map: dict[str, str] = Field(default_factory=dict)
+    in_map: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Map graph source paths to node-local input paths. Keys are paths "
+            "such as input.text, state.user.name, or context.item; values are "
+            "input fields/paths inside the node payload."
+        ),
+    )
+    input_values: dict[str, object] = Field(
+        default_factory=dict,
+        description=(
+            "Static node-local input values keyed by destination input field/path. "
+            "Use this for graph-defined constants; use in_map only for graph paths."
+        ),
+    )
+    out_map: dict[str, str] = Field(
+        default_factory=dict,
+        description=(
+            "Map node-local output paths to workflow state destinations. Keys "
+            "are output fields/paths inside the node payload; values must be "
+            "state.* destination paths."
+        ),
+    )
     retry: int | None = Field(default=None, ge=0)
     timeout_seconds: int | None = Field(default=None, gt=0)
 

src/wf_core/runtime/ops/nodes.py

@@ -33,6 +33,11 @@ def _resolve_node_execution(
     frame = run.current_frame()
     context_values = frame_context_values(frame)
     resolved_input: dict[str, Any] = {}
+    for destination_field, value in node.input_values.items():
+        try:
+            set_local_value(resolved_input, destination_field, value)
+        except LocalPathError as exc:
+            raise WorkflowExecutionError(str(exc)) from exc
     for source_path, destination_field in node.in_map.items():
         value = safe_resolve_path(
             source_path,

src/wf_core/validation/steps.py

@@ -38,6 +38,17 @@ def validate_node_use(
     state_fields = set(workflow.state_schema.fields)
     input_root_fields = set(workflow.input_schema.properties)
 
+    for destination_field in node.input_values:
+        destination_root = _local_root(destination_field)
+        if destination_root is None or (
+            destination_root != "." and destination_root not in input_fields
+        ):
+            report.add(
+                ValidationIssueCode.INVALID_NODE_INPUT_FIELD,
+                f"nodes[{index}].input_values[{destination_field!r}]",
+                f"destination field {destination_field!r} is not declared in node input schema",
+            )
+
     for source_path, destination_field in node.in_map.items():
         destination_root = _local_root(destination_field)
         if destination_root is None or (
@@ -63,6 +74,12 @@ def validate_node_use(
             f"nodes[{index}].in_map",
             "in_map has overlapping node-local input paths",
         )
+    if has_overlapping_paths([*node.input_values, *node.in_map.values()]):
+        report.add(
+            ValidationIssueCode.INVALID_NODE_INPUT_FIELD,
+            f"nodes[{index}].input_values",
+            "static input_values overlap with path-based in_map destinations",
+        )
 
     for source_field, destination_path in node.out_map.items():
         source_root = _local_root(source_field)