less dot-separated string in the Machinery

src/wf_core/models/schemas.py

@@ -98,8 +98,8 @@ class StateFieldDecl(BaseModel):
         return schema_type if isinstance(schema_type, str) else None
 
     @field_serializer("path")
-    def _serialize_path(self, path: StatePath) -> str:
-        return str(path)
+    def _serialize_path(self, path: StatePath) -> dict[str, str | list[str]]:
+        return StatePath._serialize(path)
 
     @model_validator(mode="before")
     @classmethod
@@ -147,20 +147,26 @@ class StateSchema(BaseModel):
     @property
     def fields(self) -> list[StateFieldDecl]:
         """Return the compiled field declarations for compatibility callers."""
-        return list(self.field_map().values())
+        return list(self.field_index().values())
 
-    def field_map(self) -> dict[str, StateFieldDecl]:
-        """Return reducer-aware declarations keyed by rootless dotted path."""
+    def field_index(self) -> dict[StatePath, StateFieldDecl]:
+        """Return reducer-aware declarations keyed by exact typed state path."""
         root_schema = self.model_dump(mode="json", exclude_none=True)
         return {
             path: field
             for path, field in _iter_state_field_declarations(
                 self.properties,
                 root_schema,
-                prefix="",
+                prefix=(),
             )
         }
 
+    def field_map(self) -> dict[str, StateFieldDecl]:
+        """Return reducer-aware declarations keyed by rootless dotted path."""
+        return {
+            ".".join(path.parts): field for path, field in self.field_index().items()
+        }
+
     def root_fields(self) -> set[str]:
         """Return declared top-level state field names."""
         return set(self.properties)
@@ -245,18 +251,21 @@ def _iter_state_field_declarations(
     properties: Mapping[str, Any],
     root_schema: Mapping[str, Any],
     *,
-    prefix: str,
-) -> Iterator[tuple[str, StateFieldDecl]]:
+    prefix: tuple[str, ...],
+) -> Iterator[tuple[StatePath, StateFieldDecl]]:
     for name, property_schema in properties.items():
         if not isinstance(property_schema, Mapping):
             continue
-        path = f"{prefix}.{name}" if prefix else name
+        path = StatePath((*prefix, name))
+        display_path = ".".join(path.parts)
         resolved_schema = _resolve_local_ref(property_schema, root_schema)
-        reducer = _reducer_from_property(path, property_schema)
+        reducer = _reducer_from_property(display_path, property_schema)
         trace = property_schema.get("trace", True)
         default = property_schema.get("default")
         if not isinstance(trace, bool):
-            raise ValueError(f"invalid trace for state field {path!r}: expected bool")
+            raise ValueError(
+                f"invalid trace for state field {display_path!r}: expected bool"
+            )
         validation_schema = {
             key: value
             for key, value in resolved_schema.items()
@@ -265,20 +274,22 @@ def _iter_state_field_declarations(
         _attach_root_schema_context(validation_schema, root_schema)
         yield (
             path,
-            StateFieldDecl.model_validate({
-                "path": StatePath.of(path),
-                "schema": SchemaRef.model_validate(validation_schema),
-                "reducer": reducer,
-                "trace": trace,
-                "default": default,
-            }),
+            StateFieldDecl.model_validate(
+                {
+                    "path": path,
+                    "schema": SchemaRef.model_validate(validation_schema),
+                    "reducer": reducer,
+                    "trace": trace,
+                    "default": default,
+                }
+            ),
         )
         child_properties = resolved_schema.get("properties")
         if isinstance(child_properties, Mapping):
             yield from _iter_state_field_declarations(
                 child_properties,
                 root_schema,
-                prefix=path,
+                prefix=path.parts,
             )
 
 

src/wf_core/runtime/ops/state.py

@@ -63,7 +63,7 @@ def apply_output_bindings(
             "mapped state patch has overlapping destination paths"
         )
 
-    state_fields = workflow.state_schema.field_map()
+    state_fields = workflow.state_schema.field_index()
     resolved_patch: dict[StatePath, Any] = {}
     for binding in bindings:
         try:
@@ -139,7 +139,7 @@ def write_state_value(
     validate_staged_state_patch(
         staged_state,
         {StatePath.parse(destination_path): (key_path, merged_value)},
-        workflow.state_schema.field_map(),
+        workflow.state_schema.field_index(),
     )
     state.clear()
     state.update(staged_state)
@@ -152,7 +152,7 @@ def prepare_state_value(
     value: Any,
     *,
     reducers: Mapping[str, ReducerDefinition] | None = None,
-    state_fields: Mapping[str, StateFieldDecl] | None = None,
+    state_fields: Mapping[StatePath, StateFieldDecl] | None = None,
 ) -> tuple[list[str], Any]:
     """Resolve reducer output for a state write without mutating state."""
     try:
@@ -165,9 +165,11 @@ def prepare_state_value(
             f"executor only supports writes into state.*, got {destination_path!r}"
         )
 
-    declared_path = ".".join(parts)
+    declared_path = StatePath(tuple(parts))
     fields = (
-        state_fields if state_fields is not None else workflow.state_schema.field_map()
+        state_fields
+        if state_fields is not None
+        else workflow.state_schema.field_index()
     )
     declared_field = fields.get(declared_path)
     reducer = (
@@ -194,7 +196,7 @@ def project_output(workflow: Workflow, state: dict[str, Any]) -> dict[str, Any]:
 def validate_staged_state_patch(
     staged_state: dict[str, Any],
     prepared_patch: Mapping[StatePath, tuple[list[str], Any]],
-    state_fields: Mapping[str, StateFieldDecl],
+    state_fields: Mapping[StatePath, StateFieldDecl],
 ) -> None:
     """Validate affected declared state schemas before committing a patch.
 
@@ -217,18 +219,18 @@ def validate_staged_state_patch(
 
 def _affected_state_fields(
     prepared_patch: Mapping[StatePath, tuple[list[str], Any]],
-    state_fields: Mapping[str, StateFieldDecl],
+    state_fields: Mapping[StatePath, StateFieldDecl],
 ) -> list[StateFieldDecl]:
-    affected: dict[str, StateFieldDecl] = {}
+    affected: dict[StatePath, StateFieldDecl] = {}
     for destination_path in prepared_patch:
         destination_parts = destination_path.parts
-        for key, field in state_fields.items():
+        for path, field in state_fields.items():
             field_parts = field.path.parts
             if _is_prefix(field_parts, destination_parts) or _is_prefix(
                 destination_parts,
                 field_parts,
             ):
-                affected[key] = field
+                affected[path] = field
     return sorted(
         affected.values(),
         key=lambda field: len(field.path.parts),