misc changes: returns None, root as ., add

src/wf_core/local_paths.py

@@ -10,6 +10,8 @@ class LocalPathError(ValueError):
 
 def split_local_path(path: str) -> list[str]:
     """Split one dotted node-local path, rejecting empty segments."""
+    if path == ".":
+        return []
     parts = path.split(".")
     if not path or any(not part for part in parts):
         raise LocalPathError(f"invalid local path {path!r}")
@@ -18,6 +20,8 @@ def split_local_path(path: str) -> list[str]:
 
 def get_local_value(payload: Mapping[str, Any], path: str) -> Any:
     """Resolve one node-local path from a nested mapping payload."""
+    if path == ".":
+        return dict(payload)
     current: Any = payload
     for part in split_local_path(path):
         if not isinstance(current, Mapping) or part not in current:
@@ -29,6 +33,12 @@ def get_local_value(payload: Mapping[str, Any], path: str) -> Any:
 def set_local_value(payload: dict[str, Any], path: str, value: Any) -> None:
     """Write one value into a nested node-local mapping payload."""
     parts = split_local_path(path)
+    if not parts:
+        if not isinstance(value, Mapping):
+            raise LocalPathError("root local path requires a mapping value")
+        payload.clear()
+        payload.update(value)
+        return
     current = payload
     for part in parts[:-1]:
         next_value = current.setdefault(part, {})

src/wf_core/validation/steps.py

@@ -39,11 +39,10 @@ def validate_node_use(
     input_root_fields = set(workflow.input_schema.properties)
 
     for source_path, destination_field in node.in_map.items():
-        try:
-            destination_root = split_local_path(destination_field)[0]
-        except LocalPathError:
-            destination_root = ""
-        if destination_root not in input_fields:
+        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}].in_map[{source_path!r}]",
@@ -66,11 +65,10 @@ def validate_node_use(
         )
 
     for source_field, destination_path in node.out_map.items():
-        try:
-            source_root = split_local_path(source_field)[0]
-        except LocalPathError:
-            source_root = ""
-        if source_root not in output_fields:
+        source_root = _local_root(source_field)
+        if source_root is None or (
+            source_root != "." and source_root not in output_fields
+        ):
             report.add(
                 ValidationIssueCode.INVALID_NODE_OUTPUT_FIELD,
                 f"nodes[{index}].out_map[{source_field!r}]",
@@ -90,6 +88,14 @@ def validate_node_use(
         )
 
 
+def _local_root(path: str) -> str | None:
+    try:
+        parts = split_local_path(path)
+    except LocalPathError:
+        return None
+    return "." if not parts else parts[0]
+
+
 def validate_condition_node(
     node: ConditionNode,
     index: int,