use json schema for state schema

docs/core_state_mapping_and_merge.md

@@ -160,31 +160,30 @@ creates a reusable boundary for:
 ## State Declarations and Merge Rules
 
 State merge behavior is attached to declared exact state paths. The canonical
-schema shape is a list of declarations:
+schema shape is ordinary JSON Schema. `reducer` is a wf_core extension keyword
+on property schemas; JSON Schema validators ignore it, while wf_core validates
+and uses it for state writes.
 
 ```json
 {
-  "fields": [
-    {
-      "path": "state.person.name",
-      "schema": {"type": "string"},
-      "reducer": {"name": "wf.std.replace"}
+  "type": "object",
+  "properties": {
+    "person": {
+      "type": "object",
+      "properties": {
+        "name": {"type": "string", "reducer": "wf.std.replace"},
+        "tags": {"type": "array", "reducer": "wf.std.append"}
+      }
     },
-    {
-      "path": "state.person.tags",
-      "schema": {"type": "array"},
-      "reducer": {"name": "wf.std.append"}
-    },
-    {
-      "path": "state.profile",
-      "schema": {"type": "object"},
-      "reducer": {"name": "wf.std.merge_object"}
+    "profile": {
+      "type": "object",
+      "reducer": "wf.std.merge_object"
     }
-  ]
+  }
 }
 ```
 
-Deprecated dict-shaped state fields are still accepted at parse boundaries:
+Deprecated `fields` state declarations are still accepted at parse boundaries:
 
 ```json
 {
@@ -194,14 +193,15 @@ Deprecated dict-shaped state fields are still accepted at parse boundaries:
 }
 ```
 
-Validated `StateSchema` models store and dump the canonical list shape.
-Presentation layers may rebuild a tree for humans.
+Validated `StateSchema` models store and dump the canonical JSON Schema shape.
+`StateSchema.field_map()` compiles an internal exact-path index for runtime
+reducer lookup.
 
-`wf_authoring` keeps authored schemas nested for humans and LLM clients, but
-projects nested authored state into this flat exact-path index. For example, a
-Pydantic `person: Person` field may produce declarations for `person`,
-`person.name`, and `person.tags` without forcing the author to spell those
-paths manually.
+`wf_authoring` keeps authored schemas nested for humans and LLM clients, and
+injects state metadata such as `reducer` into the generated JSON Schema
+properties. For example, a Pydantic `person: Person` field can produce nested
+properties for `person.name` and `person.tags` without forcing the author to
+spell those paths manually.
 
 ### Exact-path ownership
 

docs/path_mapping_scratch.md

@@ -139,7 +139,7 @@ Overlap rules:
 
 State path validation and write behavior:
 
-- writable `StatePath` must have its root declared in `state_schema.fields`
+- writable `StatePath` must have its root declared in `state_schema.properties`
 - whole-state write targets such as bare `state` stay out of scope for now
 - nested state subpaths are allowed once the root exists in the schema
 - exact nested state declarations are reducer/schema hints, not root ownership
@@ -476,29 +476,28 @@ Core explicitness:
 
 State schema fields:
 
-- move toward list-of-structs instead of dict keys
-- canonical shape:
+- canonical shape is normal JSON Schema
+- state field metadata such as `reducer` lives as a wf_core extension keyword on
+  each property schema
+- JSON Schema validators ignore `reducer`; wf_core validates it separately and
+  compiles it into an exact-path runtime index
+- accept old `fields` shapes at parse time for compatibility:
 
 ```text
-StateSchema.fields: list[StateFieldDecl]
-
-StateFieldDecl:
-  path: StatePath
-  type: string
-  reducer: ReducerRef
-```
-
-- serialized field paths include `state.` prefix, e.g. `state.person.tags`
-- accept old dict shape at parse time for compatibility:
-
-```text
-fields = {
-  "person.tags": {"type": "array", "reducer": "wf.std.append"}
+state_schema = {
+  "type": "object",
+  "properties": {
+    "person": {
+      "type": "object",
+      "properties": {
+        "tags": {"type": "array", "reducer": "wf.std.append"}
+      }
+    }
+  }
 }
 ```
 
-- normalize old shape to canonical list internally
-- canonical serialization emits list shape
+- canonical serialization emits JSON Schema shape
 - duplicate field paths are validation errors
 - exact reducer matching uses exact `StatePath`
 

docs/superpowers/plans/2026-05-20-json-schema-state-reducers.md

@@ -0,0 +1,249 @@
+# JSON Schema State Reducers Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Make `Workflow.state_schema` a normal JSON Schema object, with `reducer` as an explicit workflow extension keyword on field schemas.
+
+**Architecture:** `StateSchema` should validate as JSON Schema first, then expose helper indexes for workflow runtime metadata. Runtime reducer lookup should compile from `properties` paths instead of requiring a separate path declaration list. Legacy `fields` inputs remain parse-only compatibility during the transition.
+
+**Tech Stack:** Python, Pydantic v2, `jsonschema`, `wf_core` path models, pytest, basedpyright, ruff.
+
+---
+
+### Task 1: Add Canonical State Schema Tests
+
+**Files:**
+- Modify: `tests/core/test_nested_state_paths.py`
+- Modify: `tests/core/test_schema_validation.py`
+
+- [ ] **Step 1: Add a test for JSON Schema property reducers**
+
+```python
+def test_state_schema_uses_json_schema_properties_as_canonical_shape() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "type": "object",
+            "properties": {
+                "person": {
+                    "type": "object",
+                    "properties": {
+                        "name": {
+                            "type": "string",
+                            "description": "Display name",
+                            "reducer": "wf.std.replace",
+                        }
+                    },
+                },
+                "count": {"type": "integer", "reducer": "wf.std.add"},
+            },
+        }
+    )
+
+    fields = schema.field_map()
+    assert fields["person.name"].validation_schema.type == "string"
+    assert fields["person.name"].reducer == ReducerRef(name="wf.std.replace")
+    assert fields["count"].reducer == ReducerRef(name="wf.std.add")
+```
+
+- [ ] **Step 2: Add a dump test proving the canonical output is still JSON Schema**
+
+```python
+def test_state_schema_dumps_canonical_json_schema_with_reducer_keyword() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "type": "object",
+            "properties": {
+                "count": {"type": "integer", "reducer": "wf.std.add"}
+            },
+        }
+    )
+
+    dumped = schema.model_dump(mode="json")
+    assert dumped["type"] == "object"
+    assert dumped["properties"]["count"]["type"] == "integer"
+    assert dumped["properties"]["count"]["reducer"] == "wf.std.add"
+    Draft202012Validator.check_schema(dumped)
+```
+
+- [ ] **Step 3: Add a runtime reducer lookup test from canonical schema**
+
+```python
+def test_exact_nested_state_path_uses_reducer_from_json_schema_property() -> None:
+    workflow = _workflow_from_state_schema(
+        StateSchema.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "person": {
+                        "type": "object",
+                        "properties": {
+                            "tags": {"type": "array", "reducer": "wf.std.append"}
+                        },
+                    }
+                },
+            }
+        )
+    )
+    state = {"person": {"tags": ["seed"]}}
+
+    write_state_value(workflow, state, "state.person.tags", ["next"])
+
+    assert state["person"]["tags"] == ["seed", "next"]
+```
+
+- [ ] **Step 4: Run focused tests and confirm failures**
+
+Run: `uv run --with pytest pytest tests/core/test_nested_state_paths.py tests/core/test_schema_validation.py -q`
+
+Expected: new tests fail because `StateSchema` still serializes as `fields: [...]` and reducer lookup is compiled from field declarations only.
+
+### Task 2: Implement JSON-Schema-Native `StateSchema`
+
+**Files:**
+- Modify: `src/wf_core/models/schemas.py`
+
+- [ ] **Step 1: Make `StateSchema` inherit JSON Schema fields directly**
+
+`StateSchema` should expose common JSON Schema object fields:
+
+```python
+title: str | None = None
+type: str | list[str] | None = "object"
+properties: dict[str, Any] = Field(default_factory=dict)
+required: list[str] = Field(default_factory=list)
+```
+
+- [ ] **Step 2: Preserve legacy `fields` as parse-only input**
+
+Keep accepting:
+
+```json
+{"fields": [{"path": "state.count", "type": "integer", "reducer": "wf.std.add"}]}
+```
+
+and:
+
+```json
+{"fields": {"count": {"type": "integer", "reducer": "wf.std.add"}}}
+```
+
+by converting both into:
+
+```json
+{"type": "object", "properties": {"count": {"type": "integer", "reducer": "wf.std.add"}}}
+```
+
+- [ ] **Step 3: Add `field_map()` as an internal compiled index**
+
+`field_map()` should walk explicit object `properties` and return `StateFieldDecl` values keyed by rootless state path. It must:
+
+- include every explicit property path
+- parse `reducer` with `ReducerRef`
+- default missing reducer to `wf.std.replace`
+- preserve `trace` and `default` workflow extension keywords
+- remove workflow extension keywords from `StateFieldDecl.validation_schema`
+
+- [ ] **Step 4: Validate JSON Schema and extension keyword types**
+
+Use `SchemaRef`/`jsonschema` validation for the complete state schema. Add explicit validation that:
+
+- `reducer` is a string or `ReducerRef`-compatible object
+- `trace` is a boolean when present
+- `default` is allowed as JSON Schema/default metadata
+
+### Task 3: Update Artifact Reducer Extraction
+
+**Files:**
+- Modify: `src/wf_artifacts/factory.py`
+
+- [ ] **Step 1: Extract reducer dependencies from `state_schema.properties`**
+
+Add a helper that walks explicit JSON Schema properties and yields reducer payloads from every property schema.
+
+- [ ] **Step 2: Keep legacy `fields` extraction only as compatibility**
+
+If `state_schema.fields` exists in old artifacts, continue reading it. Prefer canonical `properties` when present.
+
+- [ ] **Step 3: Add tests through existing workflow surface/artifact tests**
+
+Use an existing artifact/dependency test and assert a reducer declared at:
+
+```json
+state_schema.properties.count.reducer
+```
+
+is included in required capabilities.
+
+### Task 4: Update Authoring Conversion
+
+**Files:**
+- Modify: `src/wf_authoring/schemas.py`
+- Modify: `tests/authoring/test_schemas.py`
+
+- [ ] **Step 1: Attach reducer metadata directly to generated property schemas**
+
+When `state_schema_from(BaseModel)` sees `Annotated[..., state_field(reducer=...)]`, inject `reducer` and `trace` into that property schema instead of building a separate field map.
+
+- [ ] **Step 2: Preserve model JSON Schema as the state schema**
+
+Return `StateSchema.model_validate(schema_with_reducer_keywords)` so generated state schema remains JSON Schema-shaped.
+
+### Task 5: Update Docs and Examples
+
+**Files:**
+- Modify: `docs/core_state_mapping_and_merge.md`
+- Modify: `docs/workflow_drafts.md`
+- Modify: `docs/wf_mcp_operator_manual.md`
+- Modify: `docs/wf_mcp_end_to_end_runbook.md`
+- Modify: `examples/raw_canonical_workflow.py`
+
+- [ ] **Step 1: Replace canonical `fields: [...]` examples**
+
+Use JSON Schema:
+
+```json
+{
+  "type": "object",
+  "properties": {
+    "count": {
+      "type": "integer",
+      "description": "Counter value",
+      "reducer": "wf.std.add"
+    }
+  }
+}
+```
+
+- [ ] **Step 2: Document extension semantics**
+
+State clearly that `reducer` is not standard JSON Schema behavior. JSON Schema validators ignore it; `wf_core` reads it for workflow state writes.
+
+### Task 6: Verification
+
+**Files:**
+- All touched files
+
+- [ ] **Step 1: Run focused tests**
+
+Run: `uv run --with pytest pytest tests/core/test_nested_state_paths.py tests/core/test_schema_validation.py tests/authoring/test_schemas.py -q`
+
+- [ ] **Step 2: Run full tests**
+
+Run: `uv run --with pytest pytest -q`
+
+- [ ] **Step 3: Run static checks**
+
+Run:
+
+```bash
+uvx ruff check
+uv run basedpyright --level error
+```
+
+---
+
+## Self-Review
+
+- Spec coverage: covers canonical JSON Schema state shape, reducer extension keyword, compatibility, runtime lookup, artifact dependency extraction, authoring generation, docs, and verification.
+- Placeholder scan: no placeholders remain.
+- Type consistency: `StateSchema`, `StateFieldDecl`, `ReducerRef`, and `SchemaRef` names match current code.

docs/wf_mcp_end_to_end_runbook.md

@@ -244,14 +244,13 @@ arguments:
       "required": ["text"]
     },
     "state_schema": {
-      "fields": [
-        {
-          "path": "state.echoed",
-          "schema": {
-            "type": "string"
-          }
+      "type": "object",
+      "properties": {
+        "echoed": {
+          "type": "string",
+          "reducer": "wf.std.replace"
         }
-      ]
+      }
     },
     "output_schema": {
       "type": "object",

docs/wf_mcp_operator_manual.md

@@ -389,14 +389,13 @@ Minimal example:
       "required": ["text"]
     },
     "state_schema": {
-      "fields": [
-        {
-          "path": "state.echoed",
-          "schema": {
-            "type": "string"
-          }
+      "type": "object",
+      "properties": {
+        "echoed": {
+          "type": "string",
+          "reducer": "wf.std.replace"
         }
-      ]
+      }
     },
     "output_schema": {
       "type": "object",

docs/workflow_drafts.md

@@ -48,14 +48,13 @@ A minimal draft looks like this:
     "required": ["text"]
   },
   "state_schema": {
-    "fields": [
-      {
-        "path": "state.echoed",
-        "schema": {
-          "type": "string"
-        }
+    "type": "object",
+    "properties": {
+      "echoed": {
+        "type": "string",
+        "reducer": "wf.std.replace"
       }
-    ]
+    }
   },
   "output_schema": {
     "type": "object",

examples/mcp_tool_workflow.py

@@ -42,7 +42,10 @@ async def run_example() -> dict[str, object]:
                 "properties": {"text": {"type": "string"}},
                 "required": ["text"],
             },
-            "state_schema": {"fields": {"echoed": {"type": "string"}}},
+            "state_schema": {
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+            },
             "output_schema": {
                 "type": "object",
                 "properties": {"echoed": {"type": "string"}},

examples/raw_canonical_workflow.py

@@ -15,13 +15,13 @@ def build_raw_canonical_workflow() -> Workflow:
                 "required": ["text"],
             },
             "state_schema": {
-                "fields": [
-                    {
-                        "path": "state.message",
-                        "schema": {"type": "string"},
-                        "reducer": {"name": "wf.std.replace"},
+                "type": "object",
+                "properties": {
+                    "message": {
+                        "type": "string",
+                        "reducer": "wf.std.replace",
                     }
-                ]
+                },
             },
             "output_schema": {
                 "type": "object",

src/wf_artifacts/factory.py

@@ -88,19 +88,9 @@ def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapabili
     state_schema = plan.get("state_schema")
     if not isinstance(state_schema, dict):
         return {}
-    fields = state_schema.get("fields")
-    if isinstance(fields, dict):
-        field_values = fields.values()
-    elif isinstance(fields, list):
-        field_values = fields
-    else:
-        return {}
 
     requirements: dict[str, RequiredCapability] = {}
-    for field in field_values:
-        if not isinstance(field, dict):
-            continue
-        reducer_payload = field.get("reducer", "wf.std.replace")
+    for reducer_payload in _iter_state_schema_reducer_payloads(state_schema):
         if isinstance(reducer_payload, str):
             reducer = ReducerRef(name=reducer_payload)
         else:
@@ -118,3 +108,36 @@ def _required_reducers_from_plan(plan: JsonObject) -> dict[str, RequiredCapabili
             kind="reducer",
         )
     return requirements
+
+
+def _iter_state_schema_reducer_payloads(state_schema: JsonObject) -> list[object]:
+    """Read reducer refs from canonical JSON Schema and legacy field metadata."""
+    reducer_payloads: list[object] = []
+    properties = state_schema.get("properties")
+    if isinstance(properties, dict):
+        reducer_payloads.extend(_iter_property_reducer_payloads(properties))
+
+    fields = state_schema.get("fields")
+    if isinstance(fields, dict):
+        field_values = fields.values()
+    elif isinstance(fields, list):
+        field_values = fields
+    else:
+        field_values = []
+
+    for field in field_values:
+        if isinstance(field, dict):
+            reducer_payloads.append(field.get("reducer", "wf.std.replace"))
+    return reducer_payloads
+
+
+def _iter_property_reducer_payloads(properties: JsonObject) -> list[object]:
+    payloads: list[object] = []
+    for property_schema in properties.values():
+        if not isinstance(property_schema, dict):
+            continue
+        payloads.append(property_schema.get("reducer", "wf.std.replace"))
+        child_properties = property_schema.get("properties")
+        if isinstance(child_properties, dict):
+            payloads.extend(_iter_property_reducer_payloads(child_properties))
+    return payloads

src/wf_authoring/schemas.py

@@ -6,7 +6,7 @@ from typing import Any, Iterator
 
 from pydantic import BaseModel, TypeAdapter
 
-from wf_core import ReducerRef, SchemaRef, StateField, StateSchema
+from wf_core import ReducerRef, SchemaRef, StateSchema
 
 SchemaLike = SchemaRef | type[BaseModel] | type[Any] | dict[str, Any]
 StateSchemaLike = StateSchema | type[BaseModel] | type[Any] | dict[str, Any]
@@ -47,21 +47,24 @@ def state_schema_from(value: StateSchemaLike) -> StateSchema:
     """Coerce an authoring state declaration into a core state schema."""
     if isinstance(value, StateSchema):
         return value
-    if isinstance(value, dict) and "fields" in value:
+    if isinstance(value, dict):
         return StateSchema.model_validate(value)
 
     schema = schema_ref_from(value)
+    schema_payload = schema.model_dump(mode="json", exclude_none=True)
     metadata_by_name = _state_metadata_by_name(value)
-    fields = {
-        path: StateField(
-            type=_state_field_type(property_schema),
-            reducer=metadata_by_name.get(path, StateFieldMetadata()).reducer,
-            trace=metadata_by_name.get(path, StateFieldMetadata()).trace,
-            default=_state_field_default(value, path, property_schema),
-        )
-        for path, property_schema in _flatten_state_properties(schema)
-    }
-    return StateSchema.from_field_map(fields)
+    for path, property_schema in _flatten_state_properties(schema):
+        metadata = metadata_by_name.get(path, StateFieldMetadata())
+        extension_schema = _lookup_mutable_property_schema(schema_payload, path)
+        if extension_schema is None:
+            extension_schema = property_schema
+        extension_schema["reducer"] = _dump_reducer(metadata.reducer)
+        if not metadata.trace:
+            extension_schema["trace"] = False
+        default = _state_field_default(value, path, property_schema)
+        if default is not None:
+            extension_schema["default"] = default
+    return StateSchema.model_validate(schema_payload)
 
 
 def _reducer_ref_from(value: ReducerLike) -> ReducerRef:
@@ -139,17 +142,41 @@ def _resolve_property_schema(
     return resolved if isinstance(resolved, dict) else property_schema
 
 
-def _state_field_type(property_schema: object) -> str:
-    if not isinstance(property_schema, dict):
-        return "object"
-    field_type = property_schema.get("type")
-    if isinstance(field_type, str):
-        return field_type
-    if "$ref" in property_schema or "properties" in property_schema:
-        return "object"
-    if "items" in property_schema:
-        return "array"
-    return "object"
+def _dump_reducer(reducer: ReducerRef) -> str | dict[str, Any]:
+    if not reducer.config:
+        return reducer.name
+    return reducer.model_dump(mode="json")
+
+
+def _lookup_mutable_property_schema(
+    schema: dict[str, Any],
+    path: str,
+) -> dict[str, Any] | None:
+    """Find a property schema, following local Pydantic ``$defs`` references."""
+    current: dict[str, Any] = schema
+    for part in path.split("."):
+        properties = current.get("properties")
+        if not isinstance(properties, dict):
+            return None
+        raw_child = properties.get(part)
+        if not isinstance(raw_child, dict):
+            return None
+        current = _resolve_mutable_property_schema(raw_child, schema)
+    return current
+
+
+def _resolve_mutable_property_schema(
+    property_schema: dict[str, Any],
+    root_schema: dict[str, Any],
+) -> dict[str, Any]:
+    ref = property_schema.get("$ref")
+    if not isinstance(ref, str) or not ref.startswith("#/$defs/"):
+        return property_schema
+    definitions = root_schema.get("$defs", {})
+    if not isinstance(definitions, dict):
+        return property_schema
+    resolved = definitions.get(ref.removeprefix("#/$defs/"))
+    return resolved if isinstance(resolved, dict) else property_schema
 
 
 def _state_field_default(

src/wf_core/models/schemas.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Iterator, Mapping
 from typing import Any
 
 from jsonschema import Draft202012Validator, SchemaError, validators
@@ -121,78 +121,277 @@ class StateFieldDecl(BaseModel):
 
 
 class StateSchema(BaseModel):
-    """Workflow state schema with canonical list fields.
+    """Workflow state JSON Schema plus reducer extension keywords.
 
-    Deprecated dict-shaped input is still accepted at parse time and normalized
-    so runtime and serialization only deal with list-of-struct declarations.
+    Canonical state schemas are ordinary JSON Schema objects. Field-level
+    workflow metadata such as ``reducer`` and ``trace`` lives beside JSON Schema
+    keywords inside ``properties`` entries, where JSON Schema validators will
+    ignore it and wf_core can compile it into runtime behavior.
+
+    Deprecated ``fields`` inputs are still accepted at parse time and normalized
+    into ``properties`` so persisted dumps stay JSON-Schema-shaped.
     """
 
     model_config = ConfigDict(extra="allow")
 
-    fields: list[StateFieldDecl] = Field(default_factory=list)
+    title: str | None = None
+    type: str | list[str] | None = "object"
+    properties: dict[str, Any] = Field(default_factory=dict)
+    required: list[str] = 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})
 
+    @property
+    def fields(self) -> list[StateFieldDecl]:
+        """Return the compiled field declarations for compatibility callers."""
+        return list(self.field_map().values())
+
     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}
+        """Return reducer-aware declarations keyed by rootless dotted 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="",
+            )
+        }
 
     def root_fields(self) -> set[str]:
         """Return declared top-level state field names."""
-        return {field.path.parts[0] for field in self.fields}
+        return set(self.properties)
+
+    @model_serializer(mode="wrap")
+    def _serialize_without_none_fields(self, handler: Any) -> dict[str, Any]:
+        """Persist state schemas as JSON Schema objects without null keywords."""
+        data = handler(self)
+        return {key: value for key, value in data.items() if value is not None}
 
     @model_validator(mode="before")
     @classmethod
-    def _coerce_deprecated_field_map(cls, value: object) -> object:
+    def _coerce_deprecated_fields(cls, value: object) -> object:
         if not isinstance(value, Mapping):
             return value
 
         data = dict(value)
-        fields = data.get("fields")
-        if not isinstance(fields, Mapping):
+        fields = data.pop("fields", None)
+        if fields is None:
             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(fields, list):
+            for raw_field in fields:
+                field = StateFieldDecl.model_validate(raw_field)
+                _set_state_property_schema(
+                    data,
+                    field.path.parts,
+                    _property_schema_from_field(field),
+                )
+            return data
 
+        if not isinstance(fields, Mapping):
+            raise ValueError("state_schema.fields must be a mapping or list")
+
+        for raw_path, raw_field in fields.items():
             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"
                 )
 
+            path = str(raw_path)
+            if not path.startswith("state."):
+                path = f"state.{path}"
             field_data["path"] = path
-            normalized_fields.append(field_data)
-
-        data["fields"] = normalized_fields
+            field = StateFieldDecl.model_validate(field_data)
+            _set_state_property_schema(
+                data,
+                field.path.parts,
+                _property_schema_from_field(field),
+            )
         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)
+    def _validate_state_json_schema_and_extensions(self) -> StateSchema:
+        schema = self.model_dump(mode="json", exclude_none=True)
+        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
+
+        # JSON Schema permits custom keywords, so wf_core validates reducer
+        # metadata separately instead of relying on jsonschema to reject it.
+        for path, property_schema in _iter_property_schemas(
+            self.properties,
+            schema,
+        ):
+            _validate_state_field_extensions(path, property_schema)
         return self
 
 
+def _iter_state_field_declarations(
+    properties: Mapping[str, Any],
+    root_schema: Mapping[str, Any],
+    *,
+    prefix: str,
+) -> Iterator[tuple[str, StateFieldDecl]]:
+    for name, property_schema in properties.items():
+        if not isinstance(property_schema, Mapping):
+            continue
+        path = f"{prefix}.{name}" if prefix else name
+        resolved_schema = _resolve_local_ref(property_schema, root_schema)
+        reducer = _reducer_from_property(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")
+        validation_schema = {
+            key: value
+            for key, value in resolved_schema.items()
+            if key not in {"reducer", "trace"}
+        }
+        yield (
+            path,
+            StateFieldDecl.model_validate(
+                {
+                    "path": StatePath.of(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,
+            )
+
+
+def _iter_property_schemas(
+    properties: Mapping[str, Any],
+    root_schema: Mapping[str, Any],
+    *,
+    prefix: str = "",
+) -> Iterator[tuple[str, Mapping[str, Any]]]:
+    for name, property_schema in properties.items():
+        if not isinstance(property_schema, Mapping):
+            continue
+        path = f"{prefix}.{name}" if prefix else name
+        yield path, property_schema
+        resolved_schema = _resolve_local_ref(property_schema, root_schema)
+        child_properties = resolved_schema.get("properties")
+        if isinstance(child_properties, Mapping):
+            yield from _iter_property_schemas(
+                child_properties,
+                root_schema,
+                prefix=path,
+            )
+
+
+def _validate_state_field_extensions(
+    path: str,
+    property_schema: Mapping[str, Any],
+) -> None:
+    _reducer_from_property(path, property_schema)
+    trace = property_schema.get("trace", True)
+    if not isinstance(trace, bool):
+        raise ValueError(f"invalid trace for state field {path!r}: expected bool")
+
+
+def _reducer_from_property(
+    path: str,
+    property_schema: Mapping[str, Any],
+) -> ReducerRef:
+    reducer = property_schema.get("reducer", "wf.std.replace")
+    try:
+        if isinstance(reducer, str):
+            return ReducerRef(name=reducer)
+        if isinstance(reducer, Mapping):
+            return ReducerRef.model_validate(reducer)
+    except ValueError as exc:
+        raise ValueError(f"invalid reducer for state field {path!r}: {exc}") from exc
+    raise ValueError(
+        f"invalid reducer for state field {path!r}: expected string or object"
+    )
+
+
+def _property_schema_from_field(field: StateFieldDecl) -> dict[str, Any]:
+    schema = field.validation_schema.model_dump(mode="json", exclude_none=True)
+    schema["reducer"] = _dump_reducer_keyword(field.reducer)
+    if not field.trace:
+        schema["trace"] = False
+    if field.default is not None:
+        schema["default"] = field.default
+    return schema
+
+
+def _dump_reducer_keyword(reducer: ReducerRef) -> str | dict[str, Any]:
+    if not reducer.config:
+        return reducer.name
+    return reducer.model_dump(mode="json")
+
+
+def _set_state_property_schema(
+    data: dict[str, Any],
+    path_parts: tuple[str, ...],
+    property_schema: dict[str, Any],
+) -> None:
+    data.setdefault("type", "object")
+    properties = data.setdefault("properties", {})
+    if not isinstance(properties, dict):
+        raise ValueError("state_schema.properties must be an object")
+
+    current_properties = properties
+    for part in path_parts[:-1]:
+        current = current_properties.setdefault(
+            part,
+            {"type": "object", "properties": {}},
+        )
+        if not isinstance(current, dict):
+            raise ValueError(f"state field path {'.'.join(path_parts)!r} overlaps")
+        current.setdefault("type", "object")
+        next_properties = current.setdefault("properties", {})
+        if not isinstance(next_properties, dict):
+            raise ValueError(f"state field path {'.'.join(path_parts)!r} overlaps")
+        current_properties = next_properties
+
+    leaf = path_parts[-1]
+    if leaf in current_properties:
+        raise ValueError(f"duplicate state field path {'.'.join(path_parts)!r}")
+    current_properties[leaf] = property_schema
+
+
+def _resolve_local_ref(
+    property_schema: Mapping[str, Any],
+    root_schema: Mapping[str, Any],
+) -> Mapping[str, Any]:
+    """Resolve the common Pydantic ``#/$defs/...`` case for internal indexes."""
+    ref = property_schema.get("$ref")
+    if not isinstance(ref, str) or not ref.startswith("#/$defs/"):
+        return property_schema
+    definitions = root_schema.get("$defs")
+    if not isinstance(definitions, Mapping):
+        return property_schema
+    resolved = definitions.get(ref.removeprefix("#/$defs/"))
+    return resolved if isinstance(resolved, Mapping) else property_schema
+
+
 class NodeDef(BaseModel):
     """Reusable node contract referenced by one or more node uses."""
 

tests/artifacts/test_factory.py

@@ -39,7 +39,8 @@ def test_create_workflow_artifact_from_plan_derives_boundary_schemas() -> None:
 def test_create_workflow_artifact_from_plan_adds_reducer_dependencies() -> None:
     plan = _plan()
     plan["state_schema"] = {
-        "fields": {"best_score": {"type": "integer", "reducer": "wf.std.max"}}
+        "type": "object",
+        "properties": {"best_score": {"type": "integer", "reducer": "wf.std.max"}},
     }
 
     artifact = create_workflow_artifact_from_plan(
@@ -180,7 +181,7 @@ def test_create_workflow_artifact_from_plan_rejects_missing_boundary_schema() ->
 
 def test_create_workflow_artifact_from_plan_rejects_invalid_workflow_shape() -> None:
     plan = _plan()
-    plan["state_schema"] = {"fields": {"echoed": {"schema": {"type": "string"}}}}
+    plan["state_schema"] = {"type": 123}
 
     try:
         create_workflow_artifact_from_plan(

tests/core/test_nested_state_paths.py

@@ -30,7 +30,7 @@ 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:
+def test_state_schema_accepts_legacy_field_list_and_dumps_json_schema() -> None:
     schema = StateSchema.model_validate(
         {
             "fields": [
@@ -46,6 +46,51 @@ def test_state_schema_accepts_canonical_field_list() -> None:
 
     assert schema.fields[0].path == StatePath.of("person")
     assert schema.field_map()["person.name"].type == "string"
+    dumped = schema.model_dump(mode="json")
+    assert dumped["properties"]["person"]["properties"]["name"]["type"] == "string"
+    assert "fields" not in dumped
+
+
+def test_state_schema_uses_json_schema_properties_as_canonical_shape() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "type": "object",
+            "properties": {
+                "person": {
+                    "type": "object",
+                    "properties": {
+                        "name": {
+                            "type": "string",
+                            "description": "Display name",
+                            "reducer": "wf.std.replace",
+                        }
+                    },
+                },
+                "count": {"type": "integer", "reducer": "wf.std.add"},
+            },
+        }
+    )
+
+    fields = schema.field_map()
+    assert fields["person.name"].validation_schema.type == "string"
+    assert fields["person.name"].reducer == ReducerRef(name="wf.std.replace")
+    assert fields["count"].reducer == ReducerRef(name="wf.std.add")
+
+
+def test_state_schema_rejects_invalid_reducer_extension_keyword() -> None:
+    try:
+        StateSchema.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "count": {"type": "integer", "reducer": {"bad": True}},
+                },
+            }
+        )
+    except ValueError as exc:
+        assert "invalid reducer for state field 'count'" in str(exc)
+    else:
+        raise AssertionError("expected invalid reducer extension keyword to fail")
 
 
 def test_state_schema_accepts_canonical_schema_field() -> None:
@@ -69,35 +114,28 @@ 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"
+    assert dumped["properties"]["person"]["properties"]["name"]["type"] == "string"
+    assert "fields" not in dumped
 
 
-def test_state_schema_rejects_deprecated_dict_value_with_schema_key() -> None:
-    try:
-        StateSchema.model_validate(
-            {
-                "fields": {
-                    "person.name": {
-                        "schema": {"type": "string"},
-                    }
+def test_state_schema_accepts_deprecated_dict_value_with_schema_key() -> None:
+    schema = StateSchema.model_validate(
+        {
+            "fields": {
+                "person.name": {
+                    "schema": {"type": "string", "description": "Display name"},
                 }
             }
-        )
-    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")
+        }
+    )
+
+    assert schema.field_map()["person.name"].validation_schema.type == "string"
 
 
-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_json_schema_field_without_type() -> None:
+    schema = StateSchema.model_validate({"fields": {"person.name": {"default": "Ada"}}})
+
+    assert schema.field_map()["person.name"].default == "Ada"
 
 
 def test_state_schema_accepts_deprecated_state_prefixed_dict_keys() -> None:
@@ -105,7 +143,7 @@ def test_state_schema_accepts_deprecated_state_prefixed_dict_keys() -> None:
         {"fields": {"state.person.name": {"type": "string"}}}
     )
 
-    assert schema.fields[0].path == StatePath.of("person.name")
+    assert schema.field_map()["person.name"].path == StatePath.of("person.name")
 
 
 def test_state_field_decl_model_dump_serializes_path_as_string() -> None:
@@ -122,8 +160,9 @@ def test_state_schema_model_dump_serializes_paths_as_strings() -> None:
         {"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"
+    dumped = schema.model_dump(mode="json")
+    assert dumped["properties"]["person"]["properties"]["name"]["type"] == "string"
+    assert "fields" not in dumped
 
 
 def test_state_schema_rejects_duplicate_field_paths() -> None:
@@ -142,6 +181,29 @@ def test_state_schema_rejects_duplicate_field_paths() -> None:
         raise AssertionError("expected duplicate state field path to fail")
 
 
+def test_exact_nested_state_path_uses_reducer_from_json_schema_property() -> None:
+    workflow = _workflow_from_state_schema(
+        StateSchema.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "person": {
+                        "type": "object",
+                        "properties": {
+                            "tags": {"type": "array", "reducer": "wf.std.append"}
+                        },
+                    }
+                },
+            }
+        )
+    )
+    state = {"person": {"tags": ["seed"]}}
+
+    write_state_value(workflow, state, "state.person.tags", ["next"])
+
+    assert state["person"]["tags"] == ["seed", "next"]
+
+
 def test_state_schema_field_map_uses_rootless_keys() -> None:
     schema = StateSchema.model_validate(
         {
@@ -340,10 +402,14 @@ def test_reducer_definition_can_wrap_config_aware_callable() -> None:
 
 
 def _workflow(*, fields: dict[str, StateField]) -> Workflow:
+    return _workflow_from_state_schema(StateSchema.from_field_map(fields))
+
+
+def _workflow_from_state_schema(state_schema: StateSchema) -> Workflow:
     return Workflow(
         name="nested_state_paths",
         input_schema=SchemaRef(type="object", properties={}),
-        state_schema=StateSchema.from_field_map(fields),
+        state_schema=state_schema,
         output_schema=SchemaRef(type="object", properties={}),
         node_defs=[],
         start="unused",

tests/core/test_raw_canonical_workflow_example.py

@@ -22,7 +22,7 @@ def test_raw_canonical_workflow_serializes_new_shape() -> None:
     workflow = build_raw_canonical_workflow()
     dumped = workflow.model_dump(mode="json")
     node = dumped["nodes"][0]
-    state_field = dumped["state_schema"]["fields"][0]
+    message_schema = dumped["state_schema"]["properties"]["message"]
 
     assert "input" in node
     assert "output" in node
@@ -34,5 +34,5 @@ def test_raw_canonical_workflow_serializes_new_shape() -> None:
     assert node["input"][1]["value"] == "raw:"
     assert node["output"][0]["source"] == "message"
     assert node["output"][0]["target"] == "state.message"
-    assert state_field["path"] == "state.message"
-    assert state_field["schema"]["type"] == "string"
+    assert message_schema["type"] == "string"
+    assert message_schema["reducer"] == "wf.std.replace"

tests/core/test_schema_validation.py

@@ -146,3 +146,26 @@ def test_state_field_decl_dump_omits_nested_schema_none_fields() -> None:
     assert dumped["schema"]["type"] == "object"
     assert "title" not in dumped["schema"]
     Draft202012Validator.check_schema(dumped["schema"])
+
+
+def test_state_schema_dump_is_valid_json_schema_with_reducer_keyword() -> None:
+    from wf_core import StateSchema
+
+    schema = StateSchema.model_validate(
+        {
+            "type": "object",
+            "properties": {
+                "count": {
+                    "type": "integer",
+                    "description": "Running count",
+                    "reducer": "wf.std.add",
+                }
+            },
+        }
+    )
+
+    dumped = schema.model_dump(mode="json")
+    assert dumped["type"] == "object"
+    assert dumped["properties"]["count"]["description"] == "Running count"
+    assert dumped["properties"]["count"]["reducer"] == "wf.std.add"
+    Draft202012Validator.check_schema(dumped)

tests/wf_mcp/test_workflow_surface.py

@@ -1190,7 +1190,8 @@ def _custom_reducer_artifact() -> WorkflowArtifact:
             "required": ["total", "amount"],
         },
         "state_schema": {
-            "fields": {
+            "type": "object",
+            "properties": {
                 "total": {
                     "type": "integer",
                     "reducer": "custom.multiply",