fix: delegate structured schema references

pyproject.toml

@@ -19,6 +19,7 @@ dependencies = [
     "openapi-core>=0.19",
     "pydantic>=2",
     "pyyaml>=6.0.3",
+    "referencing>=0.37",
     "typer>=0.24.2",
     "uvicorn>=0.46.0",
 ]

src/wf_api/authoring_contracts.py

@@ -8,12 +8,10 @@ from wf_core.analysis.context_scopes import (
     ContextFieldAvailability,
     context_analysis_warnings,
     context_fields_by_node,
-    normalize_definition_reference,
-    resolve_schema_reference,
-    schema_union_branches,
 )
 from wf_core.models.workflow import Workflow
 from wf_core.paths import GraphSourcePath
+from wf_core.schema_navigation import SchemaNavigator, SchemaView
 
 from .models.authoring_contracts import (
     AuthoringContractInventoryPayload,
@@ -292,105 +290,41 @@ def _nested_item_subpaths(
     *,
     depth: int,
     prefix_parts: tuple[str, ...] = (),
-    definitions: Mapping[str, Any] | None = None,
-    active_refs: frozenset[str] = frozenset(),
 ) -> list[AuthoringPathOptionPayload]:
     """Emit bounded object children beneath one foreach ``item`` schema.
 
-    Dangling ``$ref`` values resolve against the nearest enclosing ``$defs``
+    The shared navigator owns reference and composition semantics. A repeated
-    table (kept by recursive item schemas); a repeated reference stays
+    reference stays selectable at its own path but is not expanded again,
-    selectable at its own path but is not expanded again, mirroring the
+    mirroring the input/state inventory.
-    input/state inventory. Composition keywords are a union: children come
-    from every object branch, deduplicated by path.
     """
-    from .models.authoring_contracts import AuthoringPathOptionPayload as _Payload
+    return _nested_item_view_subpaths(
-
+        SchemaNavigator(item_schema).root,
-    if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH:
+        owner_id,
-        return []
+        availability,
-    table = item_schema.get("$defs")
+        depth=depth,
-    if isinstance(table, Mapping):
+        prefix_parts=prefix_parts,
-        definitions = table
+        active_references=frozenset(),
-    elif definitions is None:
+    )
-        definitions = {}
-    options: list[_Payload] = []
-    for branch in schema_union_branches(item_schema):
-        options.extend(
-            _branch_item_children(
-                branch,
-                owner_id,
-                availability,
-                depth=depth,
-                prefix_parts=prefix_parts,
-                definitions=definitions,
-                active_refs=active_refs,
-            )
-        )
-    seen: set[str] = set()
-    deduped: list[_Payload] = []
-    for option in options:
-        if option["path"] not in seen:
-            seen.add(option["path"])
-            deduped.append(option)
-    return deduped
 
 
-def _branch_item_children(
+def _nested_item_view_subpaths(
-    branch: Mapping[str, Any],
+    view: SchemaView,
     owner_id: str,
     availability: str,
     *,
     depth: int,
     prefix_parts: tuple[str, ...],
-    definitions: Mapping[str, Any],
+    active_references: frozenset[int],
-    active_refs: frozenset[str],
 ) -> list[AuthoringPathOptionPayload]:
+    """Project one resolver-aware item view into bounded authoring options."""
     from .models.authoring_contracts import AuthoringPathOptionPayload as _Payload
 
-    if isinstance(branch.get("$ref"), str):
+    if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH:
-        reference = normalize_definition_reference(branch["$ref"])
-        if reference in active_refs:
-            return []
-        resolved = resolve_schema_reference(definitions, branch)
-        if resolved is branch:
-            return []
-        return _nested_item_subpaths(
-            resolved,
-            owner_id,
-            availability,
-            depth=depth,
-            prefix_parts=prefix_parts,
-            definitions=definitions,
-            active_refs=active_refs | {reference},
-        )
-    properties = branch.get("properties")
-    if not isinstance(properties, Mapping):
         return []
     options: list[_Payload] = []
-    for name, sub_schema in properties.items():
+    for child in view.object_children(active_references=active_references):
-        if not isinstance(name, str) or not isinstance(sub_schema, Mapping):
+        name = child.name
-            continue
+        child_schema = child.view.standalone_schema()
-        if (
-            isinstance(sub_schema.get("type"), str)
-            and sub_schema.get("type") == "array"
-        ):
-            # Arrays are whole values; item indexes need real runtime indexes.
-            path = str(
-                GraphSourcePath(
-                    "context", ("foreach", owner_id, "item", *prefix_parts, name)
-                )
-            )
-            options.append(
-                {
-                    "path": path,
-                    "label": name.replace("_", " ").replace("-", " ").title(),
-                    "origin": "runtime_context",
-                    "schema": deepcopy(dict(sub_schema)),
-                    "required": False,
-                    "availability": availability,  # type: ignore[typeddict-item]
-                    "uses": ["step_input"],
-                }
-            )
-            continue
         path = str(
             GraphSourcePath(
                 "context", ("foreach", owner_id, "item", *prefix_parts, name)
@@ -401,24 +335,30 @@ def _branch_item_children(
                 "path": path,
                 "label": name.replace("_", " ").replace("-", " ").title(),
                 "origin": "runtime_context",
-                "schema": deepcopy(dict(sub_schema)),
+                "schema": child_schema,
                 "required": False,
                 "availability": availability,  # type: ignore[typeddict-item]
                 "uses": ["step_input"],
             }
         )
-        options.extend(
+        if not child.view.is_array():
-            _nested_item_subpaths(
+            options.extend(
-                sub_schema,
+                _nested_item_view_subpaths(
-                owner_id,
+                    child.view,
-                availability,
+                    owner_id,
-                depth=depth + 1,
+                    availability,
-                prefix_parts=(*prefix_parts, name),
+                    depth=depth + 1,
-                definitions=definitions,
+                    prefix_parts=(*prefix_parts, name),
-                active_refs=active_refs,
+                    active_references=child.active_references,
+                )
             )
-        )
+    seen: set[str] = set()
-    return options
+    deduped: list[_Payload] = []
+    for option in options:
+        if option["path"] not in seen:
+            seen.add(option["path"])
+            deduped.append(option)
+    return deduped
 
 
 def _context_entries_for_inventory(

src/wf_core/analysis/__init__.py

@@ -6,10 +6,7 @@ from .context_scopes import (
     context_fields_by_node,
     context_schema_for_node,
     context_schemas_by_node,
-    normalize_definition_reference,
-    resolve_schema_reference,
     root_context_schema,
-    schema_union_branches,
 )
 from .control_regions import (
     ControlRegionAnalysis,
@@ -30,8 +27,5 @@ __all__ = [
     "context_fields_by_node",
     "context_schema_for_node",
     "context_schemas_by_node",
-    "normalize_definition_reference",
-    "resolve_schema_reference",
     "root_context_schema",
-    "schema_union_branches",
 ]

src/wf_core/analysis/context_scopes.py

@@ -23,12 +23,11 @@ from wf_core.context_contracts import (
 )
 from wf_core.models.steps import ForeachNode
 from wf_core.models.workflow import Edge, Workflow
+from wf_core.schema_navigation import SchemaNavigator
 from wf_core.tokens import END
 
 type ContextAvailability = Literal["available", "conditional"]
 
-_MAX_LOCAL_SCHEMA_REFERENCE_DEPTH = 32
-
 
 @dataclass(frozen=True, slots=True)
 class ContextFieldAvailability:
@@ -357,23 +356,6 @@ def _foreach_item_schema(
     controller_stack = owner_stack_by_node.get(foreach.id)
     if controller_stack is None:
         return {}
-    source_schema = _schema_at_path(
-        workflow,
-        foreach.over.root,
-        foreach.over.parts,
-        controller_stack,
-        foreach_nodes,
-        owner_stack_by_node,
-    )
-    if not isinstance(source_schema, Mapping):
-        return {}
-    source_type = source_schema.get("type")
-    is_array = source_type == "array" or (
-        isinstance(source_type, list) and "array" in source_type
-    )
-    items = source_schema.get("items")
-    if not is_array or not isinstance(items, Mapping):
-        return {}
     document = _schema_document(
         workflow,
         foreach.over.root,
@@ -381,50 +363,13 @@ def _foreach_item_schema(
         foreach_nodes=foreach_nodes,
         owner_stack_by_node=owner_stack_by_node,
     )
-    try:
+    source = SchemaNavigator(document).at_path(foreach.over.parts)
-        resolved_items = _resolve_local_reference(document, items)
+    if source is None:
-    except ValueError:
         return {}
-    result = dict(_inline_local_refs(document, resolved_items))
+    items = source.array_items()
-    if _has_dangling_ref(result):
+    if items is None:
-        # Cut recursions keep their definitions table so downstream walkers
+        return {}
-        # can resolve through them instead of meeting a bare `$ref`.
+    return items.standalone_schema()
-        definitions = _collect_definitions(document)
-        if definitions:
-            result["$defs"] = definitions
-    return deepcopy(result)
-
-
-def _schema_at_path(
-    workflow: Workflow,
-    root: str,
-    parts: tuple[str, ...],
-    stack: ForeachOwnerStack,
-    foreach_nodes: Mapping[str, ForeachNode],
-    owner_stack_by_node: Mapping[str, ForeachOwnerStack],
-) -> Mapping[str, object] | None:
-    try:
-        schema_document = _schema_document(
-            workflow,
-            root,
-            stack=stack,
-            foreach_nodes=foreach_nodes,
-            owner_stack_by_node=owner_stack_by_node,
-        )
-        current: object = schema_document
-        for part in parts:
-            if not isinstance(current, Mapping):
-                return None
-            resolved = _resolve_local_reference(schema_document, current)
-            properties = resolved.get("properties")
-            if not isinstance(properties, Mapping):
-                return None
-            current = properties.get(part)
-        if not isinstance(current, Mapping):
-            return None
-        return _resolve_local_reference(schema_document, current)
-    except ValueError:
-        return None
 
 
 def _schema_document(
@@ -483,276 +428,3 @@ def _schema_document(
                             current[field.name] = field.schema
         return {"type": "object", "properties": current}
     return {}
-
-
-def normalize_definition_reference(reference: str) -> str:
-    """Normalize legacy ``#/definitions/`` refs to ``#/$defs/`` form."""
-    if reference.startswith("#/definitions/"):
-        return "#/$defs/" + reference.removeprefix("#/definitions/")
-    return reference
-
-
-def _merge_ref_siblings(
-    target: Mapping[str, object], node: Mapping[str, object]
-) -> dict[str, object]:
-    """Merge ``$ref`` siblings over the resolved target (2020-12 conjunction).
-
-    Scalar siblings (``description``, ``title``) override; ``properties`` union
-    per key with the sibling winning; ``required`` unions. ``$ref`` itself is
-    consumed unless the target chains to another reference.
-    """
-    merged = dict(target)
-    for key, value in node.items():
-        if key == "$ref":
-            continue
-        existing_properties = merged.get("properties")
-        if (
-            key == "properties"
-            and isinstance(value, Mapping)
-            and isinstance(existing_properties, Mapping)
-        ):
-            merged["properties"] = {**existing_properties, **value}
-            continue
-        existing_required = merged.get("required")
-        if (
-            key == "required"
-            and isinstance(value, list)
-            and isinstance(existing_required, list)
-        ):
-            merged["required"] = [
-                *existing_required,
-                *[item for item in value if item not in existing_required],
-            ]
-            continue
-        merged[key] = value
-    return merged
-
-
-def _lookup_definition(
-    definitions: Mapping[str, object], reference: str
-) -> Mapping[str, object] | None:
-    """Walk a definition pointer beneath a merged definitions table, leniently.
-
-    Only definition-table pointers resolve here; anything else returns
-    ``None`` so callers fail closed.
-    """
-    normalized = normalize_definition_reference(reference)
-    if not normalized.startswith("#/$defs/"):
-        return None
-    current: object = definitions
-    for raw_part in normalized.removeprefix("#/$defs/").split("/"):
-        part = raw_part.replace("~1", "/").replace("~0", "~")
-        if not isinstance(current, Mapping) or part not in current:
-            return None
-        current = current[part]
-    return current if isinstance(current, Mapping) else None
-
-
-def resolve_schema_reference(
-    definitions: Mapping[str, object], node: Mapping[str, object]
-) -> Mapping[str, object]:
-    """Leniently resolve one node's ``$ref`` chain against a definitions table.
-
-    Unresolvable, external, or cyclic references return ``node`` unchanged so
-    schema walkers fail closed. Sibling constraints merge like the strict
-    resolver.
-    """
-    current = node
-    seen: set[str] = set()
-    while True:
-        raw = current.get("$ref")
-        if not isinstance(raw, str):
-            return current
-        reference = normalize_definition_reference(raw)
-        if reference in seen or len(seen) >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH:
-            return node
-        seen.add(reference)
-        target = _lookup_definition(definitions, reference)
-        if target is None:
-            return node
-        current = _merge_ref_siblings(target, current)
-
-
-def schema_union_branches(node: Mapping[str, object]) -> list[Mapping[str, object]]:
-    """Return object-candidate branches: the node plus anyOf/oneOf/allOf members.
-
-    Composition keywords are a union approximation for path walking: a path is
-    readable when some branch declares it. This matches ``Optional[X]``
-    (pydantic ``anyOf``) and subclass ``allOf`` shapes; exotic intersections
-    may over-accept, which path allowlisting prefers to false rejection.
-    """
-    branches = [node]
-    for key in ("anyOf", "oneOf", "allOf"):
-        members = node.get(key)
-        if isinstance(members, list):
-            branches.extend(member for member in members if isinstance(member, Mapping))
-    return branches
-
-
-def _subtree_references(node: object) -> set[str]:
-    """Collect normalized ``$ref`` strings in a subtree (bounded scan)."""
-    found: set[str] = set()
-    seen: set[int] = set()
-    stack: list[object] = [node]
-    while stack:
-        current = stack.pop()
-        if isinstance(current, Mapping):
-            if id(current) in seen:
-                continue
-            seen.add(id(current))
-            reference = current.get("$ref")
-            if isinstance(reference, str):
-                found.add(normalize_definition_reference(reference))
-            stack.extend(current.values())
-        elif isinstance(current, list):
-            if id(current) in seen:
-                continue
-            seen.add(id(current))
-            stack.extend(current)
-    return found
-
-
-def _inline_local_refs(
-    root_schema: Mapping[str, object],
-    candidate: Mapping[str, object],
-) -> Mapping[str, object]:
-    """Return ``candidate`` with nested local refs resolved inline.
-
-    :func:`_foreach_item_schema` detaches the resolved item schema from its
-    source document, which would strand nested ``$ref`` pointers whose
-    ``$defs`` live at the document root. Inlining here resolves the
-    acyclic majority (including ``anyOf``/``allOf``/``oneOf`` composition and
-    ``$ref`` siblings) so downstream walkers mostly see plain ``properties``.
-    Cut recursions keep a normalized dangling ``$ref``; their definitions
-    table travels with the item schema (see :func:`_foreach_item_schema`) for
-    ref-aware walkers.
-    """
-
-    def inline(node: object, active: frozenset[str], depth: int) -> object:
-        if depth > _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH:
-            return node
-        if isinstance(node, list):
-            return [inline(item, active, depth + 1) for item in node]
-        if not isinstance(node, Mapping):
-            return node
-        reference = node.get("$ref")
-        if isinstance(reference, str):
-            lookup = normalize_definition_reference(reference)
-            if lookup in active:
-                rewritten = dict(node)
-                rewritten["$ref"] = lookup
-                return rewritten
-            try:
-                resolved = _resolve_local_reference(root_schema, node)
-            except ValueError:
-                rewritten = dict(node)
-                if lookup != reference:
-                    rewritten["$ref"] = lookup
-                return rewritten
-            target_refs = _subtree_references(resolved)
-            if lookup in target_refs or not target_refs.isdisjoint(active):
-                # Recursive shape: expanding would re-enter this reference or
-                # an ancestor, so keep it dangling and let the attached
-                # definitions table serve ref-aware walkers instead.
-                rewritten = dict(node)
-                rewritten["$ref"] = lookup
-                return rewritten
-            return inline(resolved, active | {lookup}, depth + 1)
-        inlined = dict(node)
-        properties = inlined.get("properties")
-        if isinstance(properties, Mapping):
-            inlined["properties"] = {
-                name: inline(sub, active, depth + 1) for name, sub in properties.items()
-            }
-        items = inlined.get("items")
-        if isinstance(items, (Mapping, list)):
-            inlined["items"] = inline(items, active, depth + 1)
-        additional = inlined.get("additionalProperties")
-        if isinstance(additional, (Mapping, list)):
-            inlined["additionalProperties"] = inline(additional, active, depth + 1)
-        prefix = inlined.get("prefixItems")
-        if isinstance(prefix, list):
-            inlined["prefixItems"] = inline(prefix, active, depth + 1)
-        for key in ("anyOf", "oneOf", "allOf"):
-            members = inlined.get(key)
-            if isinstance(members, list):
-                inlined[key] = [inline(member, active, depth + 1) for member in members]
-        return inlined
-
-    inlined = inline(candidate, frozenset(), 0)
-    if not isinstance(inlined, Mapping):
-        return candidate
-    return inlined
-
-
-def _has_dangling_ref(node: object) -> bool:
-    """Return whether any nested mapping still carries a string ``$ref``."""
-    seen: set[int] = set()
-    stack: list[object] = [node]
-    while stack:
-        current = stack.pop()
-        if isinstance(current, Mapping):
-            if id(current) in seen:
-                continue
-            seen.add(id(current))
-            if isinstance(current.get("$ref"), str):
-                return True
-            stack.extend(current.values())
-        elif isinstance(current, list):
-            if id(current) in seen:
-                continue
-            seen.add(id(current))
-            stack.extend(current)
-    return False
-
-
-def _collect_definitions(document: Mapping[str, object]) -> dict[str, object]:
-    """Merge a document's ``definitions``/``$defs`` tables (``$defs`` wins)."""
-    collected: dict[str, object] = {}
-    legacy = document.get("definitions")
-    if isinstance(legacy, Mapping):
-        collected.update(legacy)
-    modern = document.get("$defs")
-    if isinstance(modern, Mapping):
-        collected.update(modern)
-    return collected
-
-
-def _resolve_local_reference(
-    root_schema: Mapping[str, object],
-    candidate: Mapping[str, object],
-) -> Mapping[str, object]:
-    """Resolve bounded repository-local refs without becoming a full resolver.
-
-    ``$ref`` siblings merge over the resolved target (JSON Schema 2020-12
-    conjunction, bounded to scalar override plus ``properties``/``required``
-    union); the merged result keeps resolving when the target chains.
-    """
-    current = candidate
-    seen: set[str] = set()
-    while "$ref" in current:
-        reference = current["$ref"]
-        if not isinstance(reference, str):
-            raise ValueError("schema reference must be a string")
-        if reference in seen:
-            raise ValueError(f"cyclic schema reference {reference!r}")
-        if len(seen) >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH:
-            raise ValueError(
-                f"local schema reference depth exceeds "
-                f"{_MAX_LOCAL_SCHEMA_REFERENCE_DEPTH}"
-            )
-        if not (
-            reference.startswith("#/$defs/") or reference.startswith("#/definitions/")
-        ):
-            raise ValueError(f"unsupported schema reference {reference!r}")
-        seen.add(reference)
-        resolved: object = root_schema
-        for raw_part in reference.removeprefix("#/").split("/"):
-            part = raw_part.replace("~1", "/").replace("~0", "~")
-            if not isinstance(resolved, Mapping) or part not in resolved:
-                raise ValueError(f"unresolved schema reference {reference!r}")
-            resolved = resolved[part]
-        if not isinstance(resolved, Mapping):
-            raise ValueError(f"schema reference {reference!r} is not an object")
-        current = _merge_ref_siblings(resolved, current)
-    return current

src/wf_core/conditions.py

@@ -37,9 +37,7 @@ def eval_condition(
             context=context,
         )
     if isinstance(condition, NotCondition):
-        return not eval_condition(
+        return not eval_condition(condition.arg, state, workflow_input, context=context)
-            condition.arg, state, workflow_input, context=context
-        )
     if isinstance(condition, VariadicCondition):
         values = [
             eval_condition(arg, state, workflow_input, context=context)
@@ -48,9 +46,7 @@ def eval_condition(
         return all(values) if condition.op == "and" else any(values)
     if isinstance(condition, BinaryCondition):
         left = resolve_operand(condition.left, state, workflow_input, context=context)
-        right = resolve_operand(
+        right = resolve_operand(condition.right, state, workflow_input, context=context)
-            condition.right, state, workflow_input, context=context
-        )
         if condition.op == "eq":
             return left == right
         if condition.op == "ne":

src/wf_core/schema_navigation.py

@@ -0,0 +1,267 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping, Sequence
+from copy import deepcopy
+from dataclasses import dataclass
+from hashlib import sha256
+from typing import Any, Protocol
+
+from referencing import Registry, Resource
+from referencing.exceptions import Unresolvable
+from referencing.jsonschema import DRAFT202012
+
+type JsonSchema = Mapping[str, Any]
+type ReferenceIdentity = int
+
+
+class _Resolver(Protocol):
+    def lookup(self, ref: str) -> Any: ...
+
+    def in_subresource(self, subresource: Resource[Any]) -> Any: ...
+
+
+@dataclass(frozen=True, slots=True)
+class _Cursor:
+    contents: JsonSchema
+    resolver: _Resolver
+    resource_root: JsonSchema
+
+    def child(self, contents: object) -> _Cursor | None:
+        if not isinstance(contents, Mapping):
+            return None
+        resource = DRAFT202012.create_resource(contents)
+        resolver = self.resolver.in_subresource(resource)
+        resource_root = contents if resource.id() is not None else self.resource_root
+        return _Cursor(contents, resolver, resource_root)
+
+    def referenced(self) -> _Cursor | None:
+        reference = self.contents.get("$ref")
+        if not isinstance(reference, str):
+            return None
+        try:
+            resolved = self.resolver.lookup(reference)
+        except Unresolvable:
+            return None
+        if not isinstance(resolved.contents, Mapping):
+            return None
+        return _Cursor(resolved.contents, resolved.resolver, self.resource_root)
+
+
+@dataclass(frozen=True, slots=True)
+class SchemaChild:
+    """One declared property plus the reference identities used to reach it."""
+
+    name: str
+    view: SchemaView
+    active_references: frozenset[ReferenceIdentity]
+
+
+@dataclass(frozen=True, slots=True)
+class SchemaView:
+    """A resolver-aware view over one or more candidate JSON Schemas.
+
+    Multiple cursors represent structural alternatives used by workflow path
+    discovery. JSON Schema validation itself remains delegated to
+    ``jsonschema``; this view answers only the workflow-specific questions
+    "which object properties are declared?" and "what schema describes the
+    selected foreach item?".
+    """
+
+    _cursors: tuple[_Cursor, ...]
+
+    def object_children(
+        self,
+        *,
+        active_references: frozenset[ReferenceIdentity] = frozenset(),
+    ) -> tuple[SchemaChild, ...]:
+        """Return declared properties, cutting repeated refs on this walk."""
+        children: list[SchemaChild] = []
+        for cursor in self._cursors:
+            for branch, branch_references in _structural_branches(
+                cursor, active_references
+            ):
+                properties = branch.contents.get("properties")
+                if not isinstance(properties, Mapping):
+                    continue
+                for name, child_schema in properties.items():
+                    if not isinstance(name, str):
+                        continue
+                    child = branch.child(child_schema)
+                    if child is not None:
+                        children.append(
+                            SchemaChild(
+                                name,
+                                SchemaView((child,)),
+                                branch_references,
+                            )
+                        )
+        return tuple(children)
+
+    def allows_unknown_properties(self) -> bool:
+        """Return whether any structural branch is an open object schema."""
+        for cursor in self._cursors:
+            for branch, _ in _structural_branches(cursor, frozenset()):
+                schema = branch.contents
+                if schema == {} or (
+                    schema.get("type") == "object"
+                    and not isinstance(schema.get("properties"), Mapping)
+                    and schema.get("additionalProperties", True) is not False
+                ):
+                    return True
+        return False
+
+    def array_items(self) -> SchemaView | None:
+        """Return the first declared homogeneous item schema for an array."""
+        for cursor in self._cursors:
+            for branch, _ in _structural_branches(cursor, frozenset()):
+                schema_type = branch.contents.get("type")
+                is_array = schema_type == "array" or (
+                    isinstance(schema_type, list) and "array" in schema_type
+                )
+                if not is_array:
+                    continue
+                child = branch.child(branch.contents.get("items"))
+                if child is not None:
+                    return SchemaView((child,))
+        return None
+
+    def is_array(self) -> bool:
+        """Return whether any structural branch declares an array."""
+        return self.array_items() is not None
+
+    def standalone_schema(self) -> dict[str, Any]:
+        """Detach this view as a valid Draft 2020-12 schema resource.
+
+        A pure top-level ``$ref`` is replaced by its library-resolved target.
+        Sibling keywords remain a true conjunction through ``allOf``. Local
+        definition tables travel with the detached schema, and a stable
+        absolute ``$id`` makes their fragment references local to this
+        embedded resource rather than to a later enclosing context schema.
+        """
+        cursor = self._cursors[0]
+        body: dict[str, Any] = deepcopy(dict(cursor.contents))
+        reference = cursor.contents.get("$ref")
+        target = cursor.referenced()
+        if isinstance(reference, str) and target is not None:
+            siblings = {
+                key: deepcopy(value)
+                for key, value in cursor.contents.items()
+                if key != "$ref"
+            }
+            resolved = deepcopy(dict(target.contents))
+            body = {"allOf": [resolved, siblings]} if siblings else resolved
+            cursor = target
+
+        definitions = (
+            _definition_blocks(cursor.resource_root)
+            if _contains_reference(body)
+            else {}
+        )
+        if not definitions:
+            return body
+
+        identifier = _resource_identifier(body, definitions)
+        if "$id" in body or any(key in body for key in definitions):
+            return {"$id": identifier, **definitions, "allOf": [body]}
+        return {"$id": identifier, **body, **definitions}
+
+
+class SchemaNavigator:
+    """Resolve JSON Schema references while exposing workflow path semantics."""
+
+    def __init__(self, document: JsonSchema) -> None:
+        resource = DRAFT202012.create_resource(document)
+        identifier = _resource_identifier(document, {})
+        registry = Registry().with_resource(identifier, resource)
+        resolver = registry.resolver(identifier).in_subresource(resource)
+        self.root = SchemaView((_Cursor(document, resolver, document),))
+
+    def at_path(self, parts: Sequence[str]) -> SchemaView | None:
+        """Return the schema view at a declared object-property path."""
+        current = self.root
+        for part in parts:
+            matches = [
+                child.view for child in current.object_children() if child.name == part
+            ]
+            if not matches:
+                return None
+            current = SchemaView(
+                tuple(cursor for match in matches for cursor in match._cursors)
+            )
+        return current
+
+    def first_unknown(self, parts: Sequence[str]) -> tuple[str | None, str]:
+        """Return the first unknown path segment and available keys there."""
+        current = self.root
+        for part in parts:
+            children = current.object_children()
+            matches = [child.view for child in children if child.name == part]
+            if not matches:
+                if current.allows_unknown_properties():
+                    return None, ""
+                available = ",".join(sorted({child.name for child in children}))
+                return part, available
+            current = SchemaView(
+                tuple(cursor for match in matches for cursor in match._cursors)
+            )
+        return None, ""
+
+
+def _structural_branches(
+    cursor: _Cursor,
+    active_references: frozenset[ReferenceIdentity],
+) -> tuple[tuple[_Cursor, frozenset[ReferenceIdentity]], ...]:
+    """Expand refs/composition for structural discovery, not validation."""
+    branches: list[tuple[_Cursor, frozenset[ReferenceIdentity]]] = [
+        (cursor, active_references)
+    ]
+    target = cursor.referenced()
+    if target is not None:
+        identity = id(target.contents)
+        if identity not in active_references:
+            branches.extend(
+                _structural_branches(target, active_references | {identity})
+            )
+    for keyword in ("anyOf", "oneOf", "allOf"):
+        members = cursor.contents.get(keyword)
+        if not isinstance(members, list):
+            continue
+        for member in members:
+            child = cursor.child(member)
+            if child is not None:
+                branches.extend(_structural_branches(child, active_references))
+    return tuple(branches)
+
+
+def _definition_blocks(document: JsonSchema) -> dict[str, Any]:
+    blocks: dict[str, Any] = {}
+    for keyword in ("$defs", "definitions"):
+        definitions = document.get(keyword)
+        if isinstance(definitions, Mapping):
+            blocks[keyword] = deepcopy(dict(definitions))
+    return blocks
+
+
+def _contains_reference(schema: object) -> bool:
+    """Return whether a detached schema still needs a reference resource."""
+    pending = [schema]
+    while pending:
+        current = pending.pop()
+        if isinstance(current, Mapping):
+            if isinstance(current.get("$ref"), str):
+                return True
+            pending.extend(current.values())
+        elif isinstance(current, list):
+            pending.extend(current)
+    return False
+
+
+def _resource_identifier(schema: JsonSchema, definitions: Mapping[str, Any]) -> str:
+    payload = json.dumps(
+        [schema, definitions],
+        sort_keys=True,
+        separators=(",", ":"),
+        default=repr,
+    ).encode()
+    return f"urn:wf:schema:{sha256(payload).hexdigest()}"

src/wf_core/validation/context_paths.py

@@ -5,9 +5,7 @@ from typing import Any
 
 from wf_core.analysis.context_scopes import (
     ContextSchema,
-    resolve_schema_reference,
     root_context_schema,
-    schema_union_branches,
 )
 from wf_core.analysis.control_regions import ControlRegionAnalysis
 from wf_core.context_contracts import RESERVED_CONTEXT_KEYS
@@ -38,6 +36,7 @@ from wf_core.models.steps import (
 )
 from wf_core.models.workflow import Workflow
 from wf_core.paths import GraphSourcePath
+from wf_core.schema_navigation import SchemaNavigator
 from wf_core.validation.issues import ValidationIssueCode, ValidationReport
 
 
@@ -204,77 +203,12 @@ def _failing_segment(
     """Return the first unknown segment plus the keys available there.
 
     Returns ``(None, "")`` when the path walks declared properties (or
-    permissive unconstrained schemas). Dangling ``$ref`` values resolve
+    permissive unconstrained schemas). Reference resolution is delegated to
-    against the nearest enclosing ``$defs`` table (kept by recursive item
+    the shared Draft 2020-12 navigator; unresolvable refs fail closed.
-    schemas); unresolvable refs fail closed. Composition keywords are a
+    Composition is a union for path availability: a path is readable when
-    union: a path is readable when some branch declares it.
+    some structural branch declares it.
     """
-    if not parts:
+    return SchemaNavigator(schema).first_unknown(parts)
-        return None, ""
-    table = schema.get("$defs")
-    definitions = table if isinstance(table, Mapping) else {}
-    return _walk_schema(schema, parts, definitions)
-
-
-def _walk_schema(
-    node: Any,
-    parts: tuple[str, ...],
-    definitions: Mapping[str, Any],
-) -> tuple[str | None, str]:
-    """Walk one schema level: normalize, resolve refs, try union branches."""
-    if not parts:
-        return None, ""
-    if not isinstance(node, Mapping):
-        return parts[0], ""
-    table = node.get("$defs")
-    if isinstance(table, Mapping):
-        definitions = table
-    if isinstance(node.get("$ref"), str):
-        resolved = resolve_schema_reference(definitions, node)
-        if resolved is node:
-            return parts[0], ""
-        return _walk_schema(resolved, parts, definitions)
-    failures: list[tuple[str, str]] = []
-    for branch in schema_union_branches(node):
-        failing, available = _walk_branch(branch, parts, definitions)
-        if failing is None:
-            return None, ""
-        failures.append((failing, available))
-    # Prefer the failure that names available keys; single-branch schemas
-    # behave exactly as before.
-    for failing, available in failures:
-        if available:
-            return failing, available
-    return failures[0]
-
-
-def _walk_branch(
-    branch: Mapping[str, Any],
-    parts: tuple[str, ...],
-    definitions: Mapping[str, Any],
-) -> tuple[str | None, str]:
-    """Walk literal parts through one branch's declared properties."""
-    if isinstance(branch.get("$ref"), str):
-        # A referenced branch resolves first so recursion through definitions
-        # tables validates; unresolvable branches simply cannot accept.
-        resolved = resolve_schema_reference(definitions, branch)
-        if resolved is branch:
-            return parts[0], ""
-        return _walk_schema(resolved, parts, definitions)
-    part = parts[0]
-    properties = branch.get("properties")
-    if not isinstance(properties, Mapping):
-        if branch == {}:
-            return None, ""
-        if (
-            branch.get("type") == "object"
-            and branch.get("additionalProperties", True) is not False
-        ):
-            return None, ""
-        return part, ""
-    if part not in properties:
-        return part, ",".join(sorted(str(key) for key in properties))
-    return _walk_schema(properties[part], parts[1:], definitions)
 
 
 def _validate_workflow_output(workflow: Workflow, report: ValidationReport) -> None:

tests/core/test_context_scopes.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import pytest
+from jsonschema import Draft202012Validator
 
 from wf_core import END, Edge, ForeachNode, NodeUse, SchemaRef, StateSchema, Workflow
 from wf_core.analysis.context_scopes import (
@@ -471,17 +472,18 @@ def _composer_workflow() -> Workflow:
     )
 
 
-def test_ref_sibling_metadata_survives_inlining() -> None:
+def test_ref_sibling_metadata_and_properties_survive_projection() -> None:
     from wf_core.analysis.context_scopes import context_schema_for_node
+    from wf_core.schema_navigation import SchemaNavigator
 
     schema = context_schema_for_node(_composer_workflow(), "body")
-    item = schema["properties"]["foreach"]["properties"]["orders"]["properties"][
+    item = schema["properties"]["foreach"]["properties"]["orders"]["properties"]["item"]
-        "item"
-    ]
     assert set(item["properties"]) == {"sku", "nick"}
     nick = item["properties"]["nick"]
     assert nick["description"] == "Short display name"
-    assert set(nick["properties"]) == {"name", "label"}
+    navigator = SchemaNavigator(item)
+    assert navigator.first_unknown(("nick", "name")) == (None, "")
+    assert navigator.first_unknown(("nick", "label")) == (None, "")
 
 
 def test_recursive_item_schema_carries_definitions() -> None:
@@ -524,5 +526,111 @@ def test_recursive_item_schema_carries_definitions() -> None:
     )
     schema = context_schema_for_node(workflow, "body")
     item = schema["properties"]["foreach"]["properties"]["cats"]["properties"]["item"]
-    # The cut recursion keeps its definitions table instead of a bare $ref.
+    # The detached item is a resource with the definitions its refs require.
     assert item["$defs"]["Category"]["properties"]["name"] == {"type": "string"}
+
+
+@pytest.mark.parametrize(
+    ("definitions_key", "reference_prefix"),
+    [("$defs", "#/$defs/"), ("definitions", "#/definitions/")],
+)
+def test_recursive_context_schema_is_standalone(
+    definitions_key: str, reference_prefix: str
+) -> None:
+    """Recursive item refs must resolve from the complete emitted schema."""
+    from wf_core.analysis.context_scopes import context_schema_for_node
+
+    workflow = _workflow(
+        start="cats",
+        nodes=[
+            _foreach("cats", over="state.cats", alias="cat"),
+            _node("body"),
+        ],
+        edges=[
+            {"from": "cats", "outcome": "loop", "to": "body"},
+            {"from": "body", "outcome": "ok", "to": "cats"},
+            {"from": "cats", "outcome": "done", "to": END},
+        ],
+        state_schema={
+            "type": "object",
+            "properties": {
+                "cats": {
+                    "type": "array",
+                    "items": {"$ref": reference_prefix + "Category"},
+                },
+            },
+            definitions_key: {
+                "Category": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "parent": {"$ref": reference_prefix + "Category"},
+                    },
+                    "required": ["name"],
+                },
+            },
+        },
+    )
+    schema = context_schema_for_node(workflow, "body")
+    item_value = {"name": "kitten", "parent": {"name": "cat"}}
+    context_value = {
+        "prior_outcome": None,
+        "activated_incoming_edge": None,
+        "scope_id": "root",
+        "lineage_id": "item-lineage",
+        "parent_lineage_id": "root",
+        "foreach": {
+            "cats": {
+                "node_id": "cats",
+                "activation_id": "activation",
+                "frame_id": "item-frame",
+                "scope_id": "root",
+                "lineage_id": "item-lineage",
+                "index": 0,
+                "item": item_value,
+            }
+        },
+        "loop_item": item_value,
+        "loop_index": 0,
+        "cat": item_value,
+    }
+
+    Draft202012Validator.check_schema(schema)
+    assert Draft202012Validator(schema).is_valid(context_value)
+
+
+def test_ref_sibling_constraints_remain_conjunctive() -> None:
+    """A sibling keyword may tighten, but never replace, its referenced target."""
+    from wf_core.analysis.context_scopes import context_schema_for_node
+
+    workflow = _workflow(
+        start="names",
+        nodes=[
+            _foreach("names", over="state.names", alias="name"),
+            _node("body"),
+        ],
+        edges=[
+            {"from": "names", "outcome": "loop", "to": "body"},
+            {"from": "body", "outcome": "ok", "to": "names"},
+            {"from": "names", "outcome": "done", "to": END},
+        ],
+        state_schema={
+            "type": "object",
+            "properties": {
+                "names": {
+                    "type": "array",
+                    "items": {
+                        "$ref": "#/$defs/ShortName",
+                        "maxLength": 10,
+                    },
+                },
+            },
+            "$defs": {"ShortName": {"type": "string", "maxLength": 5}},
+        },
+    )
+    schema = context_schema_for_node(workflow, "body")
+    item = schema["properties"]["foreach"]["properties"]["names"]["properties"]["item"]
+    validator = Draft202012Validator(item)
+
+    assert validator.is_valid("12345")
+    assert not validator.is_valid("123456")

tests/wf_api/test_authoring_contracts.py

@@ -633,7 +633,12 @@ def _composer_inventory_workflow():
         start="orders",
         nodes=[
             ForeachNode.model_validate(
-                {"id": "orders", "type": "foreach", "over": "state.orders", "as": "order"}
+                {
+                    "id": "orders",
+                    "type": "foreach",
+                    "over": "state.orders",
+                    "as": "order",
+                }
             ),
             NodeUse(id="body", type="node", node="noop"),
         ],

uv.lock

@@ -775,6 +775,7 @@ dependencies = [
     { name = "openapi-core" },
     { name = "pydantic" },
     { name = "pyyaml" },
+    { name = "referencing" },
     { name = "typer" },
     { name = "uvicorn" },
 ]
@@ -803,6 +804,7 @@ requires-dist = [
     { name = "openapi-core", specifier = ">=0.19" },
     { name = "pydantic", specifier = ">=2" },
     { name = "pyyaml", specifier = ">=6.0.3" },
+    { name = "referencing", specifier = ">=0.37" },
     { name = "typer", specifier = ">=0.24.2" },
     { name = "uvicorn", specifier = ">=0.46.0" },
 ]