p2: old style deprecated support + wf_authoring to use new stuff

holy moly file changes

src/wf_artifacts/factory.py

@@ -89,11 +89,15 @@ def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapabili
     if not isinstance(state_schema, dict):
         return {}
     fields = state_schema.get("fields")
-    if not isinstance(fields, dict):
+    if isinstance(fields, dict):
+        field_values = fields.values()
+    elif isinstance(fields, list):
+        field_values = fields
+    else:
         return {}
 
     requirements: dict[str, RequiredCapability] = {}
-    for field in fields.values():
+    for field in field_values:
         if not isinstance(field, dict):
             continue
         reducer_payload = field.get("reducer", "wf.std.replace")

src/wf_authoring/builder/core.py

@@ -22,7 +22,14 @@ from wf_core import (
 from wf_core.errors import WorkflowExecutionError
 from wf_core.models.conditions import Condition as CoreCondition
 from wf_core.models.conditions import BinaryCondition, ExistsCondition, PathOperand
-from wf_core.models.steps import Step
+from wf_core.models.steps import (
+    InputBinding,
+    InputPathBinding,
+    InputValueBinding,
+    OutputBinding,
+    Step,
+)
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_core.runtime.ops.merges import ReducerDefinition
 
 from ..dsl import Expr, PathArg, PathExpr, compile_condition
@@ -52,14 +59,41 @@ from .refs import (
 def _condition_base(condition: CoreCondition) -> str:
     """Return a small source-derived id base when one path is obvious."""
     if isinstance(condition, ExistsCondition):
-        return slug_id(condition.path)
+        return slug_id(str(condition.path))
     if isinstance(condition, BinaryCondition) and isinstance(
         condition.left, PathOperand
     ):
-        return slug_id(condition.left.path)
+        return slug_id(str(condition.left.path))
     return "condition"
 
 
+def _canonical_input_bindings(
+    in_map: Mapping[str, str],
+    input_values: Mapping[str, Any],
+) -> list[InputBinding]:
+    """Convert authoring compatibility maps into canonical core input bindings."""
+    value_bindings = [
+        InputValueBinding(target=LocalPath.parse(target), value=value)
+        for target, value in input_values.items()
+    ]
+    path_bindings = [
+        InputPathBinding(
+            target=LocalPath.parse(target),
+            path=GraphSourcePath.parse(path),
+        )
+        for path, target in in_map.items()
+    ]
+    return [*value_bindings, *path_bindings]
+
+
+def _canonical_output_bindings(out_map: Mapping[str, str]) -> list[OutputBinding]:
+    """Convert authoring compatibility maps into canonical core output bindings."""
+    return [
+        OutputBinding(source=LocalPath.parse(source), target=StatePath.parse(target))
+        for source, target in out_map.items()
+    ]
+
+
 @dataclass(slots=True)
 class WorkflowBuilder:
     name: str
@@ -91,26 +125,31 @@ class WorkflowBuilder:
         self.node_specs[spec.name] = spec
         normalized_input_schema = cast(SchemaRef, self.input_schema)
         normalized_state_schema = cast(StateSchema, self.state_schema)
+        normalized_in_map = (
+            auto_input_map(
+                spec,
+                input_schema=normalized_input_schema,
+                state_schema=normalized_state_schema,
+            )
+            if in_map is None
+            else normalize_mapping(in_map)
+        )
+        normalized_input_values = dict(input_values or {})
+        normalized_out_map = (
+            auto_output_map(spec, state_schema=normalized_state_schema)
+            if out_map is None
+            else normalize_mapping(out_map)
+        )
         node = NodeUse(
             id=id or self._next_step_id(slug_id(spec.name)),
             type="node",
             node=spec.name,
             desc=desc or spec.description,
-            in_map=(
+            input=_canonical_input_bindings(
-                auto_input_map(
+                normalized_in_map,
-                    spec,
+                normalized_input_values,
-                    input_schema=normalized_input_schema,
-                    state_schema=normalized_state_schema,
-                )
-                if in_map is None
-                else normalize_mapping(in_map)
-            ),
-            input_values=dict(input_values or {}),
-            out_map=(
-                auto_output_map(spec, state_schema=normalized_state_schema)
-                if out_map is None
-                else normalize_mapping(out_map)
             ),
+            output=_canonical_output_bindings(normalized_out_map),
         )
         self.nodes.append(node)
         return node
@@ -132,14 +171,19 @@ class WorkflowBuilder:
         hatch for MCP/saved-workflow capability refs that are resolved later by
         the environment runner into node definitions and registry handlers.
         """
+        normalized_in_map = normalize_mapping(in_map)
+        normalized_input_values = dict(input_values or {})
+        normalized_out_map = normalize_mapping(out_map)
         node = NodeUse(
             id=id or self._next_step_id(slug_id(name)),
             type="node",
             node=name,
             desc=desc,
-            in_map=normalize_mapping(in_map),
+            input=_canonical_input_bindings(
-            input_values=dict(input_values or {}),
+                normalized_in_map,
-            out_map=normalize_mapping(out_map),
+                normalized_input_values,
+            ),
+            output=_canonical_output_bindings(normalized_out_map),
         )
         self.nodes.append(node)
         return node
@@ -326,7 +370,7 @@ class WorkflowBuilder:
         conditions: list[ConditionNode] = []
         default_target = self._resolve_branch_ref(default)
         previous_condition: ConditionNode | None = None
-        condition_base = id or slug_id(value.path)
+        condition_base = id or slug_id(str(value.path))
         for case_value, target in cases.items():
             condition = self.condition(
                 id=self._next_step_id(condition_base),

src/wf_authoring/builder/mapping.py

@@ -51,10 +51,11 @@ def auto_output_map(
     state_schema: StateSchema,
 ) -> dict[str, str]:
     """Map node output fields back into matching state fields."""
+    state_fields = state_schema.field_map()
     return {
         field: f"state.{field}"
         for field in spec.output_model.model_json_schema().get("properties", {})
-        if field in state_schema.fields
+        if field in state_fields
     }
 
 
@@ -64,7 +65,7 @@ def _auto_source_path(
     input_schema: SchemaRef,
     state_schema: StateSchema,
 ) -> str:
-    if field in state_schema.fields:
+    if field in state_schema.root_fields():
         return f"state.{field}"
     if field in input_schema.properties:
         return f"input.{field}"

src/wf_authoring/dsl/conditions.py

@@ -12,15 +12,16 @@ from wf_core.models.conditions import (
     PathOperand,
     VariadicCondition,
 )
+from wf_core.paths import GraphSourcePath
 
 from .paths import GraphPath, context_path, input_path, state_path
 
 
 def _operand(value: object) -> PathOperand | LiteralOperand:
     if isinstance(value, PathExpr):
-        return PathOperand(path=value.path)
+        return PathOperand(path=GraphSourcePath.parse(value.path))
     if isinstance(value, GraphPath):
-        return PathOperand(path=value.value)
+        return PathOperand(path=GraphSourcePath.parse(value.value))
     return LiteralOperand(value=value)
 
 
@@ -61,7 +62,7 @@ class PathExpr:
         return Expr(
             BinaryCondition(
                 op=op,
-                left=PathOperand(path=self.path),
+                left=PathOperand(path=GraphSourcePath.parse(self.path)),
                 right=_operand(other),
             )
         )
@@ -126,7 +127,9 @@ def context(field: str) -> PathExpr:
 
 
 def exists(value: PathExpr | GraphPath) -> Expr:
-    return Expr(ExistsCondition(op="exists", path=_path_str(value)))
+    return Expr(
+        ExistsCondition(op="exists", path=GraphSourcePath.parse(_path_str(value)))
+    )
 
 
 def not_(value: Condition | Expr) -> Expr:

src/wf_authoring/dsl/paths.py

@@ -2,11 +2,17 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 
+from wf_core.paths import GraphSourcePath
+
 
 @dataclass(frozen=True, slots=True)
 class GraphPath:
     value: str
 
+    def __post_init__(self) -> None:
+        """Validate authoring paths at construction so invalid roots fail early."""
+        object.__setattr__(self, "value", str(GraphSourcePath.parse(self.value)))
+
     def __str__(self) -> str:
         return self.value
 
@@ -15,13 +21,13 @@ def graph_path(value: str) -> GraphPath:
     return GraphPath(value)
 
 
-def input_path(field: str) -> GraphPath:
+def input_path(*parts: str) -> GraphPath:
-    return GraphPath(f"input.{field}")
+    return GraphPath(str(GraphSourcePath.input(*parts)))
 
 
-def state_path(field: str) -> GraphPath:
+def state_path(*parts: str) -> GraphPath:
-    return GraphPath(f"state.{field}")
+    return GraphPath(str(GraphSourcePath.state(*parts)))
 
 
-def context_path(field: str) -> GraphPath:
+def context_path(*parts: str) -> GraphPath:
-    return GraphPath(f"context.{field}")
+    return GraphPath(str(GraphSourcePath.context(*parts)))

src/wf_authoring/schemas.py

@@ -61,7 +61,7 @@ def state_schema_from(value: StateSchemaLike) -> StateSchema:
         )
         for path, property_schema in _flatten_state_properties(schema)
     }
-    return StateSchema(fields=fields)
+    return StateSchema.from_field_map(fields)
 
 
 def _reducer_ref_from(value: ReducerLike) -> ReducerRef:

src/wf_core/conditions.py

@@ -64,7 +64,7 @@ def resolve_operand(
     if isinstance(operand, LiteralOperand):
         return operand.value
     return safe_resolve_path(
-        operand.path,
+        str(operand.path),
         state=state,
         workflow_input=workflow_input,
         context={"prior_outcome": context_data},

src/wf_core/local_paths.py

@@ -49,7 +49,9 @@ def set_local_value(payload: dict[str, Any], path: str | LocalPath, value: Any)
     for part in parts[:-1]:
         next_value = current.setdefault(part, {})
         if not isinstance(next_value, dict):
-            raise LocalPathError(f"local path {str(parsed)!r} overlaps an existing value")
+            raise LocalPathError(
+                f"local path {str(parsed)!r} overlaps an existing value"
+            )
         current = next_value
     current[parts[-1]] = value
 

src/wf_core/models/conditions.py

@@ -4,11 +4,13 @@ from typing import Annotated, Any, Literal
 
 from pydantic import BaseModel, Field
 
+from wf_core.paths import GraphSourcePath
+
 
 class PathOperand(BaseModel):
     """Condition operand resolved from a workflow graph path."""
 
-    path: str
+    path: GraphSourcePath
 
 
 class LiteralOperand(BaseModel):
@@ -25,7 +27,7 @@ class ExistsCondition(BaseModel):
     """Condition that is true when a graph path resolves to a present value."""
 
     op: Literal["exists"]
-    path: str
+    path: GraphSourcePath
 
 
 class NotCondition(BaseModel):

src/wf_core/models/schemas.py

@@ -1,22 +1,60 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from jsonschema import Draft202012Validator, SchemaError, validators
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    field_serializer,
+    field_validator,
+    model_serializer,
+    model_validator,
+)
 
 from wf_core.models.reducers import ReducerRef
+from wf_core.paths import StatePath
 
 
 class SchemaRef(BaseModel):
-    """JSON-schema-like shape used at workflow boundaries."""
+    """JSON Schema object used at workflow and node boundaries."""
 
     model_config = ConfigDict(extra="allow")
 
     title: str | None = None
-    type: str | None = None
+    type: str | list[str] | None = None
     properties: dict[str, Any] = Field(default_factory=dict)
     required: list[str] = Field(default_factory=list)
 
+    @model_serializer(mode="wrap")
+    def _serialize_without_none_fields(self, handler: Any) -> dict[str, Any]:
+        """Persist JSON Schema objects without null-valued optional keywords."""
+        data = handler(self)
+        return {key: value for key, value in data.items() if value is not None}
+
+    @model_validator(mode="before")
+    @classmethod
+    def _validate_json_schema_definition(cls, value: object) -> object:
+        if isinstance(value, SchemaRef):
+            schema = value.model_dump(mode="json", exclude_none=True)
+        elif isinstance(value, Mapping):
+            schema = dict(value)
+        else:
+            return value
+
+        validator_cls = (
+            validators.validator_for(schema)
+            if "$schema" in schema
+            else Draft202012Validator
+        )
+        try:
+            validator_cls.check_schema(schema)
+        except SchemaError as exc:
+            raise ValueError(f"invalid JSON Schema: {exc.message}") from exc
+        return value
+
 
 class StateField(BaseModel):
     """Declared state path plus its runtime merge behavior."""
@@ -36,12 +74,123 @@ class StateField(BaseModel):
         return value
 
 
+class StateFieldDecl(BaseModel):
+    """One declared state path plus validation and reducer metadata."""
+
+    model_config = ConfigDict(populate_by_name=True, serialize_by_alias=True)
+
+    path: StatePath
+    validation_schema: SchemaRef = Field(
+        default_factory=lambda: SchemaRef(type="object"),
+        alias="schema",
+        serialization_alias="schema",
+    )
+    reducer: ReducerRef = Field(
+        default_factory=lambda: ReducerRef(name="wf.std.replace")
+    )
+    trace: bool = True
+    default: Any = None
+
+    @property
+    def type(self) -> str | None:
+        """Compatibility accessor for callers migrating from StateField.type."""
+        schema_type = self.validation_schema.type
+        return schema_type if isinstance(schema_type, str) else None
+
+    @field_serializer("path")
+    def _serialize_path(self, path: StatePath) -> str:
+        return str(path)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_legacy_type(cls, value: object) -> object:
+        if not isinstance(value, Mapping):
+            return value
+
+        data = dict(value)
+        if "schema" not in data and "type" in data:
+            data["schema"] = {"type": data.pop("type")}
+        return data
+
+    @field_validator("reducer", mode="before")
+    @classmethod
+    def _coerce_reducer(cls, value: object) -> object:
+        if isinstance(value, str):
+            return {"name": value}
+        return value
+
+
 class StateSchema(BaseModel):
-    """Workflow state schema keyed by declared exact state path."""
+    """Workflow state schema with canonical list fields.
+
+    Deprecated dict-shaped input is still accepted at parse time and normalized
+    so runtime and serialization only deal with list-of-struct declarations.
+    """
 
     model_config = ConfigDict(extra="allow")
 
-    fields: dict[str, StateField] = Field(default_factory=dict)
+    fields: list[StateFieldDecl] = Field(default_factory=list)
+
+    @classmethod
+    def from_field_map(cls, fields: Mapping[str, StateField]) -> StateSchema:
+        """Build from the deprecated dict shape at typed Python call sites."""
+        return cls.model_validate({"fields": fields})
+
+    def field_map(self) -> dict[str, StateFieldDecl]:
+        """Return declarations keyed by rootless dotted path."""
+        return {".".join(field.path.parts): field for field in self.fields}
+
+    def root_fields(self) -> set[str]:
+        """Return declared top-level state field names."""
+        return {field.path.parts[0] for field in self.fields}
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_deprecated_field_map(cls, value: object) -> object:
+        if not isinstance(value, Mapping):
+            return value
+
+        data = dict(value)
+        fields = data.get("fields")
+        if not isinstance(fields, Mapping):
+            return data
+
+        normalized_fields: list[object] = []
+        for raw_path, raw_field in fields.items():
+            path = str(raw_path)
+            if not path.startswith("state."):
+                path = f"state.{path}"
+
+            if isinstance(raw_field, BaseModel):
+                field_data = raw_field.model_dump(mode="python")
+            elif isinstance(raw_field, Mapping):
+                field_data = dict(raw_field)
+                if "schema" in field_data or "type" not in field_data:
+                    raise ValueError(
+                        "legacy state field map entries must include 'type'; "
+                        "use canonical list form for entries with 'schema'"
+                    )
+            else:
+                raise ValueError(
+                    "legacy state field map entries must include 'type'; "
+                    "use canonical list form for non-legacy declarations"
+                )
+
+            field_data["path"] = path
+            normalized_fields.append(field_data)
+
+        data["fields"] = normalized_fields
+        return data
+
+    @model_validator(mode="after")
+    def _reject_duplicate_field_paths(self) -> StateSchema:
+        seen: set[str] = set()
+        for field in self.fields:
+            key = ".".join(field.path.parts)
+            if key in seen:
+                raise ValueError(f"duplicate state field path {key!r}")
+            seen.add(key)
+        return self
 
 
 class NodeDef(BaseModel):

src/wf_core/models/steps.py

@@ -78,14 +78,15 @@ class NodeUse(BaseModel):
         input_values = cls._deprecated_mapping(
             normalized.pop("input_values", {}), field_name="input_values"
         )
-        in_map = cls._deprecated_mapping(normalized.pop("in_map", {}), field_name="in_map")
+        in_map = cls._deprecated_mapping(
+            normalized.pop("in_map", {}), field_name="in_map"
+        )
         out_map = cls._deprecated_mapping(
             normalized.pop("out_map", {}), field_name="out_map"
         )
 
         input_bindings.extend(
-            {"target": target, "value": value}
+            {"target": target, "value": value} for target, value in input_values.items()
-            for target, value in input_values.items()
         )
         input_bindings.extend(
             {"target": target, "path": path} for path, target in in_map.items()

src/wf_core/paths.py

@@ -90,7 +90,7 @@ class LocalPath:
         return core_schema.no_info_plain_validator_function(
             validate,
             serialization=core_schema.plain_serializer_function_ser_schema(
-                str, when_used="json"
+                str,
             ),
         )
 
@@ -112,9 +112,7 @@ class GraphSourcePath:
     parts: tuple[str, ...] = ()
 
     _ROOTS: ClassVar[set[str]] = {"input", "state", "context"}
-    _JSON_PATTERN: ClassVar[str] = (
+    _JSON_PATTERN: ClassVar[str] = r"^(input|state|context)(\.[A-Za-z_][A-Za-z0-9_]*)*$"
-        r"^(input|state|context)(\.[A-Za-z_][A-Za-z0-9_]*)*$"
-    )
 
     def __post_init__(self) -> None:
         if self.root not in self._ROOTS:
@@ -123,8 +121,7 @@ class GraphSourcePath:
             self,
             "parts",
             tuple(
-                _validate_segment(part, path_kind="graph source")
+                _validate_segment(part, path_kind="graph source") for part in self.parts
-                for part in self.parts
             ),
         )
 
@@ -167,7 +164,7 @@ class GraphSourcePath:
         return core_schema.no_info_plain_validator_function(
             validate,
             serialization=core_schema.plain_serializer_function_ser_schema(
-                str, when_used="json"
+                str,
             ),
         )
 
@@ -187,7 +184,9 @@ class StatePath:
 
     parts: tuple[str, ...]
 
-    _JSON_PATTERN: ClassVar[str] = r"^state\.[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$"
+    _JSON_PATTERN: ClassVar[str] = (
+        r"^state\.[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$"
+    )
 
     def __post_init__(self) -> None:
         parts = tuple(_validate_segment(part, path_kind="state") for part in self.parts)
@@ -226,7 +225,7 @@ class StatePath:
         return core_schema.no_info_plain_validator_function(
             validate,
             serialization=core_schema.plain_serializer_function_ser_schema(
-                str, when_used="json"
+                str,
             ),
         )
 

src/wf_core/runtime/ops/runs.py

@@ -3,15 +3,15 @@ from __future__ import annotations
 from copy import deepcopy
 
 from wf_core.models.workflow import Workflow
+from wf_core.paths import set_nested_value
 from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, RunStatus
 
 
 def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> RunState:
-    state = {
+    state: dict[str, object] = {}
-        name: deepcopy(field.default)
+    for field in workflow.state_schema.fields:
-        for name, field in workflow.state_schema.fields.items()
+        if field.default is not None:
-        if field.default is not None
+            set_nested_value(state, list(field.path.parts), deepcopy(field.default))
-    }
     state.update(dict(workflow_input))
     run = RunState(
         workflow_name=workflow.name,

src/wf_core/runtime/ops/state.py

@@ -7,6 +7,7 @@ from typing import Any
 from wf_core.errors import WorkflowExecutionError
 from wf_core.local_paths import LocalPathError, get_local_value, has_overlapping_paths
 from wf_core.models.reducers import ReducerRef
+from wf_core.models.schemas import StateFieldDecl
 from wf_core.models.steps import NodeUse, OutputBinding
 from wf_core.models.workflow import Workflow
 from wf_core.paths import (
@@ -59,6 +60,7 @@ def apply_output_bindings(
             "mapped state patch has overlapping destination paths"
         )
 
+    state_fields = workflow.state_schema.field_map()
     resolved_patch: dict[StatePath, Any] = {}
     for binding in bindings:
         try:
@@ -77,6 +79,7 @@ def apply_output_bindings(
             destination_path,
             value,
             reducers=reducers,
+            state_fields=state_fields,
         )
         prepared_patch[destination_path] = (key_path, merged_value)
 
@@ -137,6 +140,7 @@ def prepare_state_value(
     value: Any,
     *,
     reducers: Mapping[str, ReducerDefinition] | None = None,
+    state_fields: Mapping[str, StateFieldDecl] | None = None,
 ) -> tuple[list[str], Any]:
     """Resolve reducer output for a state write without mutating state."""
     try:
@@ -150,7 +154,10 @@ def prepare_state_value(
         )
 
     declared_path = ".".join(parts)
-    declared_field = workflow.state_schema.fields.get(declared_path)
+    fields = (
+        state_fields if state_fields is not None else workflow.state_schema.field_map()
+    )
+    declared_field = fields.get(declared_path)
     reducer = (
         declared_field.reducer if declared_field else ReducerRef(name="wf.std.replace")
     )

src/wf_core/validation/core.py

@@ -54,7 +54,7 @@ def _validate_nodes(
     report: ValidationReport,
 ) -> dict[str, Step]:
     nodes_by_id: dict[str, Step] = {}
-    state_root_fields = set(workflow.state_schema.fields)
+    state_root_fields = workflow.state_schema.root_fields()
     input_root_fields = set(workflow.input_schema.properties)
 
     for index, node in enumerate(workflow.nodes):

src/wf_core/validation/steps.py

@@ -47,8 +47,7 @@ def validate_node_use(
 
     input_fields = set(node_def.input_schema.properties)
     output_fields = set(node_def.output_schema.properties)
-    state_fields = set(workflow.state_schema.fields)
+    state_root_fields = workflow.state_schema.root_fields()
-    state_root_fields = {field.split(".", maxsplit=1)[0] for field in state_fields}
     input_root_fields = set(workflow.input_schema.properties)
 
     input_targets = []

tests/artifacts/test_draft_adapter.py

@@ -6,6 +6,7 @@ from wf_artifacts.drafts import WorkflowDraft
 from wf_artifacts.drafts.api import compile_workflow_draft, validate_workflow_draft
 from wf_artifacts.drafts.adapter import build_workflow_from_draft
 from wf_core import ConditionNode, NodeUse
+from wf_core.models.steps import InputValueBinding
 
 
 def test_adapter_lowers_keyed_use_steps_and_routes_through_builder() -> None:
@@ -56,8 +57,10 @@ def test_adapter_lowers_static_inputs_for_constant_like_steps() -> None:
 
     assert isinstance(node, NodeUse)
     assert node.node == "wf.std.constant"
-    assert node.input_values["value"] == "CLICKED"
+    assert len(node.input) == 1
-    assert node.in_map == {}
+    assert isinstance(node.input[0], InputValueBinding)
+    assert str(node.input[0].target) == "value"
+    assert node.input[0].value == "CLICKED"
 
 
 def test_invalid_literal_input_map_does_not_fall_through_to_join() -> None:

tests/authoring/test_builder.py

@@ -4,6 +4,8 @@ import pytest
 
 from wf_authoring import WorkflowBuilder, state
 from wf_core import END, RunStatus, WorkflowExecutionError
+from wf_core.models.steps import InputPathBinding
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
 from tests.authoring.helpers import (
     AutoBindInput,
@@ -28,14 +30,16 @@ def test_builder_auto_binds_matching_node_inputs_and_outputs_to_state() -> None:
         {"text": "hello", "count": 1},
     )
 
-    assert step.in_map == {
+    assert isinstance(step.input[0], InputPathBinding)
-        "state.text": "text",
+    assert step.input[0].path == GraphSourcePath.state("text")
-        "state.count": "count",
+    assert step.input[0].target == LocalPath.of("text")
-    }
+    assert isinstance(step.input[1], InputPathBinding)
-    assert step.out_map == {
+    assert step.input[1].path == GraphSourcePath.state("count")
-        "text": "state.text",
+    assert step.input[1].target == LocalPath.of("count")
-        "count": "state.count",
+    assert step.output[0].source == LocalPath.of("text")
-    }
+    assert step.output[0].target == StatePath.of("text")
+    assert step.output[1].source == LocalPath.of("count")
+    assert step.output[1].target == StatePath.of("count")
     assert run.status == RunStatus.COMPLETED
     assert run.state["text"] == "HELLO"
     assert run.state["count"] == 2
@@ -55,8 +59,11 @@ def test_builder_preserves_explicit_nested_node_local_maps() -> None:
         out_map={"payload.text": "state.text"},
     )
 
-    assert step.in_map == {"state.text": "payload.text"}
+    assert isinstance(step.input[0], InputPathBinding)
-    assert step.out_map == {"payload.text": "state.text"}
+    assert step.input[0].path == GraphSourcePath.state("text")
+    assert step.input[0].target == LocalPath.of("payload.text")
+    assert step.output[0].source == LocalPath.of("payload.text")
+    assert step.output[0].target == StatePath.of("text")
 
 
 def test_builder_preserves_explicit_root_node_local_maps() -> None:
@@ -73,8 +80,38 @@ def test_builder_preserves_explicit_root_node_local_maps() -> None:
         out_map={".": "state.text"},
     )
 
-    assert step.in_map == {"state.text": "."}
+    assert isinstance(step.input[0], InputPathBinding)
-    assert step.out_map == {".": "state.text"}
+    assert step.input[0].path == GraphSourcePath.state("text")
+    assert step.input[0].target == LocalPath.root()
+    assert step.output[0].source == LocalPath.root()
+    assert step.output[0].target == StatePath.of("text")
+
+
+def test_builder_emits_canonical_node_bindings() -> None:
+    builder = WorkflowBuilder(
+        name="canonical_bindings",
+        input_schema=AutoBindInput,
+        state_schema=AutoBindState,
+        output_schema=AutoBindOutput,
+        start="update",
+    )
+    step = builder.use(
+        auto_bind_node,
+        id="update",
+        in_map={"input.text": "text"},
+        out_map={"text": "state.text"},
+    )
+    builder.connect(step, "ok", END)
+
+    dumped_node = builder.compile().model_dump(mode="json")["nodes"][0]
+
+    assert dumped_node["input"][0]["path"] == "input.text"
+    assert dumped_node["input"][0]["target"] == "text"
+    assert dumped_node["output"][0]["source"] == "text"
+    assert dumped_node["output"][0]["target"] == "state.text"
+    assert "in_map" not in dumped_node
+    assert "input_values" not in dumped_node
+    assert "out_map" not in dumped_node
 
 
 def test_builder_can_auto_id_node_uses_from_spec_name() -> None:
@@ -208,6 +245,9 @@ def test_builder_use_ref_creates_external_node_use_without_node_def() -> None:
     workflow = builder.compile()
 
     assert step.node == "demo.echo"
-    assert step.in_map["input.text"] == "text"
+    assert isinstance(step.input[0], InputPathBinding)
-    assert step.out_map["echoed"] == "state.echoed"
+    assert step.input[0].path == GraphSourcePath.input("text")
+    assert step.input[0].target == LocalPath.of("text")
+    assert step.output[0].source == LocalPath.of("echoed")
+    assert step.output[0].target == StatePath.of("echoed")
     assert workflow.node_defs == []

tests/authoring/test_conditions.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from wf_authoring import exists, expr, not_, state, state_path
 from wf_core.conditions import eval_condition
+from wf_core.models.conditions import BinaryCondition, ExistsCondition, PathOperand
+from wf_core.paths import GraphSourcePath
 
 
 def test_condition_dsl_compiles_to_core_condition() -> None:
@@ -9,20 +11,27 @@ def test_condition_dsl_compiles_to_core_condition() -> None:
         state_path("summary")
     )
 
-    assert condition.to_condition().model_dump() == {
+    dumped = condition.to_condition().model_dump(mode="json")
-        "op": "and",
+
-        "args": [
+    assert dumped["op"] == "and"
-            {
+    assert dumped["args"][0]["left"]["path"] == "state.should_email"
-                "op": "eq",
+    assert dumped["args"][0]["right"]["value"] is True
-                "left": {"path": "state.should_email"},
+    assert dumped["args"][1]["path"] == "state.summary"
-                "right": {"value": True},
+
-            },
+
-            {
+def test_condition_dsl_compiles_authoring_paths_to_typed_core_paths() -> None:
-                "op": "exists",
+    comparison = state("score").gt(expr(state_path("threshold"))).to_condition()
-                "path": "state.summary",
+    existence = exists(state_path("summary")).to_condition()
-            },
+
-        ],
+    assert isinstance(comparison, BinaryCondition)
-    }
+    assert isinstance(comparison.left, PathOperand)
+    assert isinstance(comparison.right, PathOperand)
+    assert comparison.left.path == GraphSourcePath.state("score")
+    assert comparison.right.path == GraphSourcePath.state("threshold")
+    assert comparison.model_dump(mode="json")["left"]["path"] == "state.score"
+    assert isinstance(existence, ExistsCondition)
+    assert existence.path == GraphSourcePath.state("summary")
+    assert existence.model_dump(mode="json")["path"] == "state.summary"
 
 
 def test_condition_dsl_supports_not_ge_and_ne() -> None:

tests/authoring/test_ops.py

@@ -40,8 +40,8 @@ def _build_first_workflow(use_safe_first: bool = False):
     builder = WorkflowBuilder(
         name="first_demo",
         input_schema=SchemaRef(type="object"),
-        state_schema=StateSchema(
+        state_schema=StateSchema.from_field_map(
-            fields={
+            {
                 "items": StateField(type="array"),
                 "item": StateField(type="object"),
             }
@@ -63,8 +63,8 @@ def _build_first_maybe_workflow():
     builder = WorkflowBuilder(
         name="first_maybe_demo",
         input_schema=SchemaRef(type="object"),
-        state_schema=StateSchema(
+        state_schema=StateSchema.from_field_map(
-            fields={
+            {
                 "items": StateField(type="array"),
                 "item": StateField(type="object"),
                 "missing": StateField(type="boolean"),

tests/authoring/test_schemas.py

@@ -26,10 +26,11 @@ def test_builder_accepts_basemodel_classes_for_workflow_schemas() -> None:
 
     assert workflow.input_schema.properties["text"]["type"] == "string"
     assert workflow.output_schema.properties["text"]["type"] == "string"
-    assert set(workflow.state_schema.fields) == {"text", "count", "tags"}
+    fields = workflow.state_schema.field_map()
-    assert workflow.state_schema.fields["text"].type == "string"
+    assert set(fields) == {"text", "count", "tags"}
-    assert workflow.state_schema.fields["count"].type == "integer"
+    assert fields["text"].type == "string"
-    assert workflow.state_schema.fields["tags"].type == "array"
+    assert fields["count"].type == "integer"
+    assert fields["tags"].type == "array"
 
 
 def test_builder_accepts_typeddict_for_json_schema_refs() -> None:
@@ -57,8 +58,9 @@ def test_state_basemodel_can_declare_reducer_with_annotated_metadata() -> None:
 
     workflow = builder.compile()
 
-    assert workflow.state_schema.fields["items"].type == "array"
+    fields = workflow.state_schema.field_map()
-    assert workflow.state_schema.fields["items"].reducer.name == "wf.std.append"
+    assert fields["items"].type == "array"
+    assert fields["items"].reducer.name == "wf.std.append"
 
 
 def test_state_basemodel_seeds_safe_initial_defaults() -> None:
@@ -72,9 +74,10 @@ def test_state_basemodel_seeds_safe_initial_defaults() -> None:
 
     workflow = builder.compile()
 
-    assert workflow.state_schema.fields["items"].default == []
+    fields = workflow.state_schema.field_map()
-    assert workflow.state_schema.fields["metadata"].default == {}
+    assert fields["items"].default == []
-    assert workflow.state_schema.fields["explicit"].default == 3
+    assert fields["metadata"].default == {}
+    assert fields["explicit"].default == 3
 
 
 def test_nested_state_basemodel_projects_parent_and_child_paths() -> None:
@@ -88,13 +91,14 @@ def test_nested_state_basemodel_projects_parent_and_child_paths() -> None:
 
     workflow = builder.compile()
 
-    assert set(workflow.state_schema.fields) == {
+    fields = workflow.state_schema.field_map()
+    assert set(fields) == {
         "person",
         "person.name",
         "person.tags",
     }
-    assert workflow.state_schema.fields["person"].type == "object"
+    assert fields["person"].type == "object"
-    assert workflow.state_schema.fields["person.name"].type == "string"
+    assert fields["person.name"].type == "string"
-    assert workflow.state_schema.fields["person.tags"].type == "array"
+    assert fields["person.tags"].type == "array"
-    assert workflow.state_schema.fields["person"].reducer.name == "wf.std.replace"
+    assert fields["person"].reducer.name == "wf.std.replace"
-    assert workflow.state_schema.fields["person.tags"].reducer.name == "wf.std.append"
+    assert fields["person.tags"].reducer.name == "wf.std.append"

tests/core/test_atomic_state_patches.py

@@ -138,8 +138,8 @@ def _workflow(
     return Workflow(
         name="patch",
         input_schema=SchemaRef(type="object", properties={}),
-        state_schema=StateSchema(
+        state_schema=StateSchema.from_field_map(
-            fields=fields
+            fields
             or {
                 "person": StateField(type="object"),
                 "person.name": StateField(type="string"),
@@ -157,8 +157,12 @@ def _workflow_with_node() -> Workflow:
     return Workflow(
         name="canonical_output",
         input_schema=SchemaRef(type="object", properties={}),
-        state_schema=StateSchema(fields={"person.name": StateField(type="string")}),
+        state_schema=StateSchema.from_field_map(
-        output_schema=SchemaRef(type="object", properties={"person": {"type": "object"}}),
+            {"person.name": StateField(type="string")}
+        ),
+        output_schema=SchemaRef(
+            type="object", properties={"person": {"type": "object"}}
+        ),
         node_defs=[
             NodeDef(
                 name="rename",
@@ -177,7 +181,9 @@ def _workflow_with_node() -> Workflow:
                     "id": "rename",
                     "type": "node",
                     "node": "rename",
-                    "output": [{"source": "person.name", "target": "state.person.name"}],
+                    "output": [
+                        {"source": "person.name", "target": "state.person.name"}
+                    ],
                 }
             )
         ],

tests/core/test_canonical_node_bindings.py

@@ -57,6 +57,30 @@ def test_node_use_converts_old_maps_to_canonical_bindings():
     assert dumped["output"][0]["target"] == "state.echoed"
 
 
+def test_node_use_serializes_canonical_binding_paths_as_strings_in_all_dump_modes():
+    node = NodeUse.model_validate(
+        {
+            "id": "echo",
+            "type": "node",
+            "node": "echo",
+            "input": [{"target": "message", "path": "input.message"}],
+            "output": [{"source": "echoed", "target": "state.echoed"}],
+        }
+    )
+
+    python_dumped = node.model_dump()
+    json_dumped = node.model_dump(mode="json")
+
+    assert python_dumped["input"][0]["target"] == "message"
+    assert python_dumped["input"][0]["path"] == "input.message"
+    assert python_dumped["output"][0]["source"] == "echoed"
+    assert python_dumped["output"][0]["target"] == "state.echoed"
+    assert json_dumped["input"][0]["target"] == "message"
+    assert json_dumped["input"][0]["path"] == "input.message"
+    assert json_dumped["output"][0]["source"] == "echoed"
+    assert json_dumped["output"][0]["target"] == "state.echoed"
+
+
 def test_node_use_rejects_mixed_old_and_new_binding_styles():
     with pytest.raises(ValidationError):
         NodeUse.model_validate(
@@ -77,9 +101,7 @@ def test_input_binding_rejects_path_and_value_together():
                 "id": "bad",
                 "type": "node",
                 "node": "bad",
-                "input": [
+                "input": [{"target": "message", "path": "input.message", "value": "x"}],
-                    {"target": "message", "path": "input.message", "value": "x"}
-                ],
             }
         )
 

tests/core/test_mapping_validation.py

@@ -62,7 +62,9 @@ def test_validation_rejects_invalid_canonical_input_source_path() -> None:
     )
 
 
-def test_validation_allows_canonical_input_source_under_declared_state_field_root() -> None:
+def test_validation_allows_canonical_input_source_under_declared_state_field_root() -> (
+    None
+):
     report = _workflow(
         input=[{"target": "user.name", "path": "state.person.name"}],
         output=[],
@@ -179,8 +181,8 @@ def _workflow(
         input_schema=SchemaRef.model_validate(
             {"type": "object", "properties": {"person": {"type": "object"}}}
         ),
-        state_schema=StateSchema(
+        state_schema=StateSchema.from_field_map(
-            fields=state_fields or {"person": StateField(type="object")}
+            state_fields or {"person": StateField(type="object")}
         ),
         output_schema=SchemaRef(type="object", properties={}),
         node_defs=[

tests/core/test_nested_mappings.py

@@ -136,7 +136,7 @@ def test_root_node_local_paths_map_whole_input_and_output_payloads() -> None:
                 "properties": {"rates": {"type": "object"}},
             }
         ),
-        state_schema=StateSchema(fields={"rates": StateField(type="object")}),
+        state_schema=StateSchema.from_field_map({"rates": StateField(type="object")}),
         output_schema=SchemaRef(type="object", properties={}),
         node_defs=[
             NodeDef(
@@ -191,7 +191,7 @@ def test_static_input_values_are_merged_into_node_local_input() -> None:
     workflow = Workflow(
         name="static_input_values",
         input_schema=SchemaRef.model_validate({"type": "object", "properties": {}}),
-        state_schema=StateSchema(fields={"message": StateField(type="string")}),
+        state_schema=StateSchema.from_field_map({"message": StateField(type="string")}),
         output_schema=SchemaRef(type="object", properties={}),
         node_defs=[
             NodeDef(
@@ -253,8 +253,8 @@ def _nested_mapping_workflow() -> Workflow:
                 },
             }
         ),
-        state_schema=StateSchema(
+        state_schema=StateSchema.from_field_map(
-            fields={
+            {
                 "person": StateField(type="object"),
                 "experience": StateField(type="object"),
             }

tests/core/test_nested_state_paths.py

@@ -8,7 +8,10 @@ from wf_core import (
     StateSchema,
     Workflow,
 )
+from wf_core.models.schemas import StateFieldDecl
+from wf_core.paths import StatePath
 from wf_core.runtime.ops.merges import ReducerDefinition, apply_reducer
+from wf_core.runtime.ops.runs import create_run_state
 from wf_core.runtime.ops.state import write_state_value
 
 
@@ -27,6 +30,146 @@ def test_exact_nested_state_path_uses_declared_reducer() -> None:
     assert state["person"]["tags"] == ["seed", "next"]
 
 
+def test_state_schema_accepts_canonical_field_list() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "fields": [
+                {"path": "state.person", "type": "object"},
+                {
+                    "path": "state.person.name",
+                    "type": "string",
+                    "reducer": "wf.std.replace",
+                },
+            ]
+        }
+    )
+
+    assert schema.fields[0].path == StatePath.of("person")
+    assert schema.field_map()["person.name"].type == "string"
+
+
+def test_state_schema_accepts_canonical_schema_field() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "fields": [
+                {
+                    "path": "state.person.name",
+                    "schema": {"type": "string", "title": "Person Name"},
+                }
+            ]
+        }
+    )
+
+    field = schema.field_map()["person.name"]
+    assert field.validation_schema.type == "string"
+    assert field.validation_schema.title == "Person Name"
+
+
+def test_state_schema_accepts_deprecated_dict_shape_and_dumps_list() -> None:
+    schema = StateSchema.model_validate({"fields": {"person.name": {"type": "string"}}})
+
+    dumped = schema.model_dump(mode="json")
+    assert dumped["fields"][0]["path"] == "state.person.name"
+
+
+def test_state_schema_rejects_deprecated_dict_value_with_schema_key() -> None:
+    try:
+        StateSchema.model_validate(
+            {
+                "fields": {
+                    "person.name": {
+                        "schema": {"type": "string"},
+                    }
+                }
+            }
+        )
+    except ValueError as exc:
+        assert "legacy state field map entries must include 'type'" in str(exc)
+        assert "use canonical list form" in str(exc)
+    else:
+        raise AssertionError("expected legacy state field schema key to fail")
+
+
+def test_state_schema_rejects_deprecated_dict_value_without_type() -> None:
+    try:
+        StateSchema.model_validate({"fields": {"person.name": {"default": "Ada"}}})
+    except ValueError as exc:
+        assert "legacy state field map entries must include 'type'" in str(exc)
+        assert "use canonical list form" in str(exc)
+    else:
+        raise AssertionError("expected legacy state field without type to fail")
+
+
+def test_state_schema_accepts_deprecated_state_prefixed_dict_keys() -> None:
+    schema = StateSchema.model_validate(
+        {"fields": {"state.person.name": {"type": "string"}}}
+    )
+
+    assert schema.fields[0].path == StatePath.of("person.name")
+
+
+def test_state_field_decl_model_dump_serializes_path_as_string() -> None:
+    field = StateFieldDecl.model_validate(
+        {"path": "state.person.name", "type": "string"}
+    )
+
+    assert field.model_dump()["path"] == "state.person.name"
+    assert field.model_dump(mode="json")["path"] == "state.person.name"
+
+
+def test_state_schema_model_dump_serializes_paths_as_strings() -> None:
+    schema = StateSchema.model_validate(
+        {"fields": [{"path": "state.person.name", "type": "string"}]}
+    )
+
+    assert schema.model_dump()["fields"][0]["path"] == "state.person.name"
+    assert schema.model_dump(mode="json")["fields"][0]["path"] == "state.person.name"
+
+
+def test_state_schema_rejects_duplicate_field_paths() -> None:
+    try:
+        StateSchema.model_validate(
+            {
+                "fields": [
+                    {"path": "state.person.name", "type": "string"},
+                    {"path": "state.person.name", "type": "string"},
+                ]
+            }
+        )
+    except ValueError as exc:
+        assert "duplicate state field path 'person.name'" in str(exc)
+    else:
+        raise AssertionError("expected duplicate state field path to fail")
+
+
+def test_state_schema_field_map_uses_rootless_keys() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "fields": [
+                {"path": "state.person.name", "type": "string"},
+                {"path": "state.person.tags", "type": "array"},
+            ]
+        }
+    )
+
+    fields = schema.field_map()
+    assert fields["person.name"].path == StatePath.of("person.name")
+    assert fields["person.tags"].type == "array"
+
+
+def test_create_run_state_writes_nested_defaults_by_state_path() -> None:
+    workflow = _workflow(
+        fields={
+            "person.name": StateField(type="string", default="Ada"),
+        }
+    )
+
+    run = create_run_state(workflow, {})
+
+    assert run.state["person"]["name"] == "Ada"
+    assert "person.name" not in run.state
+
+
 def test_parent_state_declaration_does_not_apply_to_nested_write() -> None:
     workflow = _workflow(
         fields={
@@ -179,8 +322,9 @@ def test_reducer_definition_can_wrap_config_aware_callable() -> None:
                 "additionalProperties": False,
             },
         ),
-        fn=lambda current, incoming, config: ((current or 0) + incoming)
+        fn=lambda current, incoming, config: (
-        % config["modulus"],
+            ((current or 0) + incoming) % config["modulus"]
+        ),
         accepts_config=True,
     )
 
@@ -199,7 +343,7 @@ def _workflow(*, fields: dict[str, StateField]) -> Workflow:
     return Workflow(
         name="nested_state_paths",
         input_schema=SchemaRef(type="object", properties={}),
-        state_schema=StateSchema(fields=fields),
+        state_schema=StateSchema.from_field_map(fields),
         output_schema=SchemaRef(type="object", properties={}),
         node_defs=[],
         start="unused",

tests/core/test_path_values.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import pytest
 from pydantic import BaseModel, ValidationError
 
+from wf_core.models.conditions import PathOperand
 from wf_core.paths import (
     GraphSourcePath,
     LocalPath,
@@ -96,7 +97,9 @@ def test_path_objects_are_immutable_and_hashable() -> None:
         (LocalPath, (("items[0]",),)),
     ],
 )
-def test_direct_constructors_enforce_path_invariants(factory, args: tuple[object, ...]) -> None:
+def test_direct_constructors_enforce_path_invariants(
+    factory, args: tuple[object, ...]
+) -> None:
     with pytest.raises(PathResolutionError):
         factory(*args)
 
@@ -118,11 +121,29 @@ def test_pydantic_revalidates_existing_path_objects() -> None:
     object.__setattr__(local, "parts", ("items[0]",))
 
     with pytest.raises(ValidationError):
-        Payload.model_validate({"source": source, "target": StatePath.of("person"), "local": LocalPath.root()})
+        Payload.model_validate(
+            {
+                "source": source,
+                "target": StatePath.of("person"),
+                "local": LocalPath.root(),
+            }
+        )
     with pytest.raises(ValidationError):
-        Payload.model_validate({"source": GraphSourcePath.input("user"), "target": target, "local": LocalPath.root()})
+        Payload.model_validate(
+            {
+                "source": GraphSourcePath.input("user"),
+                "target": target,
+                "local": LocalPath.root(),
+            }
+        )
     with pytest.raises(ValidationError):
-        Payload.model_validate({"source": GraphSourcePath.input("user"), "target": StatePath.of("person"), "local": local})
+        Payload.model_validate(
+            {
+                "source": GraphSourcePath.input("user"),
+                "target": StatePath.of("person"),
+                "local": local,
+            }
+        )
 
 
 def test_pydantic_accepts_path_strings_and_serializes_strings() -> None:
@@ -144,6 +165,18 @@ def test_pydantic_accepts_path_strings_and_serializes_strings() -> None:
     assert dumped["target"] == "state.person"
     assert dumped["local"] == "user"
 
+    python_dumped = payload.model_dump()
+    assert python_dumped["source"] == "input.user"
+    assert python_dumped["target"] == "state.person"
+    assert python_dumped["local"] == "user"
+
+
+def test_condition_path_operand_serializes_path_as_string_in_all_dump_modes() -> None:
+    operand = PathOperand.model_validate({"path": "state.x"})
+
+    assert operand.model_dump()["path"] == "state.x"
+    assert operand.model_dump(mode="json")["path"] == "state.x"
+
 
 def test_pydantic_accepts_existing_path_objects() -> None:
     class Payload(BaseModel):

tests/core/test_schema_validation.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 import pytest
+from jsonschema import Draft202012Validator
+from pydantic import ValidationError
 
 from wf_core import SchemaRef, WorkflowExecutionError
+from wf_core.models.schemas import StateFieldDecl
 from wf_core.runtime.ops.schemas import validate_payload_against_schema
 
 
@@ -61,3 +64,85 @@ def test_schema_validation_accepts_valid_payload() -> None:
     )
 
     validate_payload_against_schema(schema, {"tags": ["a", "b"]}, "node input")
+
+
+def test_schema_ref_accepts_and_preserves_schema_with_defs_and_ref() -> None:
+    schema = SchemaRef.model_validate(
+        {
+            "$schema": "https://json-schema.org/draft/2020-12/schema",
+            "$defs": {
+                "tag": {
+                    "type": "object",
+                    "properties": {"name": {"type": "string"}},
+                    "required": ["name"],
+                }
+            },
+            "type": "object",
+            "properties": {"tag": {"$ref": "#/$defs/tag"}},
+            "required": ["tag"],
+        }
+    )
+
+    dumped = schema.model_dump(mode="json")
+
+    assert dumped["$schema"] == "https://json-schema.org/draft/2020-12/schema"
+    assert dumped["$defs"]["tag"]["type"] == "object"
+    assert dumped["properties"]["tag"]["$ref"] == "#/$defs/tag"
+    assert dumped["required"] == ["tag"]
+
+
+def test_schema_ref_rejects_invalid_json_schema_shape() -> None:
+    with pytest.raises(ValidationError, match="invalid JSON Schema"):
+        SchemaRef.model_validate({"type": 123})
+
+
+def test_schema_ref_defaults_to_draft_2020_12_without_schema_keyword() -> None:
+    schema = SchemaRef.model_validate(
+        {
+            "type": "object",
+            "properties": {"count": {"type": "integer"}},
+            "required": ["count"],
+        }
+    )
+
+    dumped = schema.model_dump(mode="json")
+
+    assert "$schema" not in dumped
+    assert dumped["type"] == "object"
+    assert dumped["properties"]["count"]["type"] == "integer"
+
+
+def test_schema_ref_preserves_extra_json_schema_keywords() -> None:
+    schema = SchemaRef.model_validate(
+        {
+            "type": "object",
+            "properties": {"name": {"type": "string"}},
+            "additionalProperties": False,
+        }
+    )
+
+    dumped = schema.model_dump(mode="json")
+
+    assert dumped["type"] == "object"
+    assert dumped["additionalProperties"] is False
+    assert dumped["properties"]["name"]["type"] == "string"
+
+
+def test_schema_ref_dump_omits_none_fields_and_stays_valid_json_schema() -> None:
+    dumped = SchemaRef(type="object").model_dump(mode="json")
+
+    assert "title" not in dumped
+    assert dumped["type"] == "object"
+    Draft202012Validator.check_schema(dumped)
+
+
+def test_state_field_decl_dump_omits_nested_schema_none_fields() -> None:
+    field = StateFieldDecl.model_validate(
+        {"path": "state.person", "schema": {"type": "object"}}
+    )
+
+    dumped = field.model_dump(mode="json")
+
+    assert dumped["schema"]["type"] == "object"
+    assert "title" not in dumped["schema"]
+    Draft202012Validator.check_schema(dumped["schema"])