diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py
index 3765f9ed..cd11b064 100644
--- a/src/wf_api/authoring_contracts.py
+++ b/src/wf_api/authoring_contracts.py
@@ -8,6 +8,9 @@ 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
@@ -289,13 +292,77 @@ 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."""
+    """Emit bounded object children beneath one foreach ``item`` schema.
+
+    Dangling ``$ref`` values resolve against the nearest enclosing ``$defs``
+    table (kept by recursive item schemas); a repeated reference stays
+    selectable at its own path but is not expanded again, mirroring the
+    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
 
     if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH:
         return []
-    properties = item_schema.get("properties")
+    table = item_schema.get("$defs")
+    if isinstance(table, Mapping):
+        definitions = table
+    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(
+    branch: Mapping[str, Any],
+    owner_id: str,
+    availability: str,
+    *,
+    depth: int,
+    prefix_parts: tuple[str, ...],
+    definitions: Mapping[str, Any],
+    active_refs: frozenset[str],
+) -> list[AuthoringPathOptionPayload]:
+    from .models.authoring_contracts import AuthoringPathOptionPayload as _Payload
+
+    if isinstance(branch.get("$ref"), str):
+        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] = []
@@ -347,6 +414,8 @@ def _nested_item_subpaths(
                 availability,
                 depth=depth + 1,
                 prefix_parts=(*prefix_parts, name),
+                definitions=definitions,
+                active_refs=active_refs,
             )
         )
     return options
diff --git a/src/wf_core/analysis/__init__.py b/src/wf_core/analysis/__init__.py
index 03c82910..3002def1 100644
--- a/src/wf_core/analysis/__init__.py
+++ b/src/wf_core/analysis/__init__.py
@@ -6,7 +6,10 @@ 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,
@@ -27,5 +30,8 @@ __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",
 ]
diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py
index e8e05872..7858a299 100644
--- a/src/wf_core/analysis/context_scopes.py
+++ b/src/wf_core/analysis/context_scopes.py
@@ -385,7 +385,14 @@ def _foreach_item_schema(
         resolved_items = _resolve_local_reference(document, items)
     except ValueError:
         return {}
-    return deepcopy(dict(_inline_local_refs(document, resolved_items)))
+    result = dict(_inline_local_refs(document, resolved_items))
+    if _has_dangling_ref(result):
+        # Cut recursions keep their definitions table so downstream walkers
+        # can resolve through them instead of meeting a bare `$ref`.
+        definitions = _collect_definitions(document)
+        if definitions:
+            result["$defs"] = definitions
+    return deepcopy(result)
 
 
 def _schema_at_path(
@@ -478,6 +485,133 @@ def _schema_document(
     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],
@@ -486,11 +620,12 @@ def _inline_local_refs(
 
     :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 keeps every downstream
-    schema walker (validation, authoring inventory) working on plain
-    ``properties`` without threading definition tables through per-node
-    schemas. Cyclic or otherwise unresolvable refs are left in place:
-    downstream walkers already treat a bare ``$ref`` as fail-closed.
+    ``$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:
@@ -502,13 +637,27 @@ def _inline_local_refs(
             return node
         reference = node.get("$ref")
         if isinstance(reference, str):
-            if reference in active:
-                return node
+            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:
-                return node
-            return inline(resolved, active | {reference}, depth + 1)
+                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):
@@ -524,6 +673,10 @@ def _inline_local_refs(
         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)
@@ -532,11 +685,49 @@ def _inline_local_refs(
     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."""
+    """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:
@@ -563,5 +754,5 @@ def _resolve_local_reference(
             resolved = resolved[part]
         if not isinstance(resolved, Mapping):
             raise ValueError(f"schema reference {reference!r} is not an object")
-        current = resolved
+        current = _merge_ref_siblings(resolved, current)
     return current
diff --git a/src/wf_core/validation/context_paths.py b/src/wf_core/validation/context_paths.py
index 9b328aa0..8a4d4d7d 100644
--- a/src/wf_core/validation/context_paths.py
+++ b/src/wf_core/validation/context_paths.py
@@ -3,7 +3,12 @@ from __future__ import annotations
 from collections.abc import Iterator, Mapping
 from typing import Any
 
-from wf_core.analysis.context_scopes import ContextSchema, root_context_schema
+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
 from wf_core.models.conditions import (
@@ -199,32 +204,77 @@ 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). A bare ``$ref`` fails closed: generated
-    per-node schemas are inline except for cyclic shapes, which cannot be
-    proven valid statically.
+    permissive unconstrained schemas). Dangling ``$ref`` values resolve
+    against the nearest enclosing ``$defs`` table (kept by recursive item
+    schemas); unresolvable refs fail closed. Composition keywords are a
+    union: a path is readable when some branch declares it.
     """
     if not parts:
         return None, ""
-    current: Any = schema
-    for part in parts:
-        if not isinstance(current, Mapping):
-            return part, ""
-        while isinstance(current.get("$ref"), str):
-            return part, ""
-        properties = current.get("properties")
-        if not isinstance(properties, Mapping):
-            if current == {}:
-                return None, ""
-            if (
-                current.get("type") == "object"
-                and current.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))
-        current = properties[part]
-    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:
diff --git a/tests/core/test_context_scopes.py b/tests/core/test_context_scopes.py
index 9ac8979b..3af9a9f2 100644
--- a/tests/core/test_context_scopes.py
+++ b/tests/core/test_context_scopes.py
@@ -424,3 +424,105 @@ def test_scoped_cycle_terminates_and_preserves_scoped_field_availability() -> No
     # A canonical back-edge pops the item stack, so the controller itself
     # stays in the outer region and exposes no item alias.
     assert "item" not in _field_map(workflow, "each")
+
+
+def _composer_state_schema() -> dict[str, object]:
+    return {
+        "type": "object",
+        "properties": {
+            "orders_list": {
+                "type": "array",
+                "items": {"$ref": "#/$defs/Order"},
+            },
+        },
+        "$defs": {
+            "Order": {
+                "type": "object",
+                "properties": {
+                    "sku": {"type": "string"},
+                    "nick": {
+                        "$ref": "#/$defs/Detail",
+                        "description": "Short display name",
+                        "properties": {"label": {"type": "string"}},
+                    },
+                },
+            },
+            "Detail": {
+                "type": "object",
+                "properties": {"name": {"type": "string"}},
+            },
+        },
+    }
+
+
+def _composer_workflow() -> Workflow:
+    return _workflow(
+        start="orders",
+        nodes=[
+            _foreach("orders", over="state.orders_list", alias="order"),
+            _node("body"),
+        ],
+        edges=[
+            {"from": "orders", "outcome": "loop", "to": "body"},
+            {"from": "body", "outcome": "ok", "to": "orders"},
+            {"from": "orders", "outcome": "done", "to": END},
+        ],
+        state_schema=_composer_state_schema(),
+    )
+
+
+def test_ref_sibling_metadata_survives_inlining() -> None:
+    from wf_core.analysis.context_scopes import context_schema_for_node
+
+    schema = context_schema_for_node(_composer_workflow(), "body")
+    item = schema["properties"]["foreach"]["properties"]["orders"]["properties"][
+        "item"
+    ]
+    assert set(item["properties"]) == {"sku", "nick"}
+    nick = item["properties"]["nick"]
+    assert nick["description"] == "Short display name"
+    assert set(nick["properties"]) == {"name", "label"}
+
+
+def test_recursive_item_schema_carries_definitions() -> None:
+    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": "#/$defs/Category"},
+                },
+            },
+            "$defs": {
+                "Category": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "parent": {
+                            "anyOf": [
+                                {"$ref": "#/$defs/Category"},
+                                {"type": "null"},
+                            ]
+                        },
+                    },
+                },
+            },
+        },
+    )
+    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.
+    assert item["$defs"]["Category"]["properties"]["name"] == {"type": "string"}
diff --git a/tests/core/test_structured_context_validation.py b/tests/core/test_structured_context_validation.py
index 98107e47..0260d529 100644
--- a/tests/core/test_structured_context_validation.py
+++ b/tests/core/test_structured_context_validation.py
@@ -592,3 +592,143 @@ def test_nested_ref_item_unknown_leaf_reports_available_keys() -> None:
     assert issue is not None
     assert "'bogus'" in issue.message
     assert "available: name" in issue.message
+
+
+def _composer_ref_workflow(*, work_path: str) -> Workflow:
+    """State schema with anyOf-optional, sibling-extended, recursive shapes."""
+    workflow = _base_workflow(work_path=work_path)
+    workflow.state_schema = StateSchema.model_validate(
+        {
+            "type": "object",
+            "properties": {
+                "items": {"type": "array", "items": {"type": "string"}},
+                "orders_list": {
+                    "type": "array",
+                    "items": {"$ref": "#/$defs/Order"},
+                },
+            },
+            "$defs": {
+                "Order": {
+                    "type": "object",
+                    "properties": {
+                        "sku": {"type": "string"},
+                        "detail": {
+                            "anyOf": [
+                                {"$ref": "#/$defs/Detail"},
+                                {"type": "null"},
+                            ]
+                        },
+                        "nick": {
+                            "$ref": "#/$defs/Detail",
+                            "description": "Short display name",
+                            "properties": {"label": {"type": "string"}},
+                        },
+                    },
+                    "required": ["sku"],
+                },
+                "Detail": {
+                    "type": "object",
+                    "properties": {"name": {"type": "string"}},
+                    "required": ["name"],
+                },
+            },
+        }
+    )
+    return workflow
+
+
+def test_optional_nested_model_subpath_is_accepted() -> None:
+    from wf_core.validation import validate_workflow
+
+    workflow = _composer_ref_workflow(
+        work_path="context.foreach.orders.item.detail.name"
+    )
+    report = validate_workflow(workflow)
+    assert [
+        issue
+        for issue in report.errors
+        if issue.code == ValidationIssueCode.INVALID_CONTEXT_PATH
+    ] == []
+
+
+def test_ref_sibling_properties_extend_the_target() -> None:
+    from wf_core.validation import validate_workflow
+
+    for work_path in (
+        "context.foreach.orders.item.nick.name",
+        "context.foreach.orders.item.nick.label",
+    ):
+        workflow = _composer_ref_workflow(work_path=work_path)
+        report = validate_workflow(workflow)
+        assert [
+            issue
+            for issue in report.errors
+            if issue.code == ValidationIssueCode.INVALID_CONTEXT_PATH
+        ] == []
+
+
+def _recursive_ref_workflow(*, work_path: str) -> Workflow:
+    workflow = _base_workflow(work_path=work_path)
+    workflow.state_schema = StateSchema.model_validate(
+        {
+            "type": "object",
+            "properties": {
+                "items": {"type": "array", "items": {"type": "string"}},
+                "orders_list": {
+                    "type": "array",
+                    "items": {"$ref": "#/$defs/Category"},
+                },
+            },
+            "$defs": {
+                "Category": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "parent": {
+                            "anyOf": [
+                                {"$ref": "#/$defs/Category"},
+                                {"type": "null"},
+                            ]
+                        },
+                        "children": {
+                            "type": "array",
+                            "items": {"$ref": "#/$defs/Category"},
+                        },
+                    },
+                    "required": ["name"],
+                },
+            },
+        }
+    )
+    return workflow
+
+
+def test_recursive_ref_paths_validate_through_definitions() -> None:
+    from wf_core.validation import validate_workflow
+
+    for work_path in (
+        "context.foreach.orders.item.parent.name",
+        "context.foreach.orders.item.children",
+    ):
+        workflow = _recursive_ref_workflow(work_path=work_path)
+        report = validate_workflow(workflow)
+        assert [
+            issue
+            for issue in report.errors
+            if issue.code == ValidationIssueCode.INVALID_CONTEXT_PATH
+        ] == []
+
+
+def test_recursive_ref_unknown_leaf_reports_defined_keys() -> None:
+    from wf_core.validation import validate_workflow
+
+    workflow = _recursive_ref_workflow(
+        work_path="context.foreach.orders.item.parent.bogus"
+    )
+    report = validate_workflow(workflow)
+    issue = _issue(
+        report, ValidationIssueCode.INVALID_CONTEXT_PATH, "nodes[2].input[0].path"
+    )
+    assert issue is not None
+    assert "'bogus'" in issue.message
+    assert "available: children,name,parent" in issue.message
diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py
index ef66b759..68c20484 100644
--- a/tests/wf_api/test_authoring_contracts.py
+++ b/tests/wf_api/test_authoring_contracts.py
@@ -586,3 +586,134 @@ def test_root_inventory_offers_foreach_map() -> None:
     paths = {option["path"] for option in options}
     assert "context.foreach" in paths
     assert "context.prior_outcome" in paths
+
+
+def _composer_inventory_workflow():
+    from wf_core import END, Edge, ForeachNode, NodeUse, SchemaRef, Workflow
+    from wf_core.models.schemas import StateSchema
+
+    return Workflow(
+        name="inventory_composer_refs",
+        input_schema=SchemaRef(type="object"),
+        state_schema=StateSchema.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "orders": {
+                        "type": "array",
+                        "items": {"$ref": "#/$defs/Order"},
+                    },
+                },
+                "$defs": {
+                    "Order": {
+                        "type": "object",
+                        "properties": {
+                            "sku": {"type": "string"},
+                            "detail": {
+                                "anyOf": [
+                                    {"$ref": "#/$defs/Detail"},
+                                    {"type": "null"},
+                                ]
+                            },
+                            "nick": {
+                                "$ref": "#/$defs/Detail",
+                                "description": "Short display name",
+                                "properties": {"label": {"type": "string"}},
+                            },
+                        },
+                    },
+                    "Detail": {
+                        "type": "object",
+                        "properties": {"name": {"type": "string"}},
+                    },
+                },
+            }
+        ),
+        output_schema=SchemaRef(type="object"),
+        start="orders",
+        nodes=[
+            ForeachNode.model_validate(
+                {"id": "orders", "type": "foreach", "over": "state.orders", "as": "order"}
+            ),
+            NodeUse(id="body", type="node", node="noop"),
+        ],
+        edges=[
+            Edge.model_validate({"from": "orders", "outcome": "loop", "to": "body"}),
+            Edge.model_validate({"from": "body", "outcome": "ok", "to": "orders"}),
+            Edge.model_validate({"from": "orders", "outcome": "done", "to": END}),
+        ],
+    )
+
+
+def test_composer_ref_children_appear_in_authoring_inventory() -> None:
+    from wf_api.authoring_contracts import context_path_options_for_node
+
+    options = context_path_options_for_node(_composer_inventory_workflow(), "body")
+    paths = {option["path"] for option in options}
+    assert "context.foreach.orders.item.detail.name" in paths
+    assert "context.foreach.orders.item.nick.name" in paths
+    assert "context.foreach.orders.item.nick.label" in paths
+
+
+def _recursive_inventory_workflow():
+    from wf_core import END, Edge, ForeachNode, NodeUse, SchemaRef, Workflow
+    from wf_core.models.schemas import StateSchema
+
+    return Workflow(
+        name="inventory_recursive_refs",
+        input_schema=SchemaRef(type="object"),
+        state_schema=StateSchema.model_validate(
+            {
+                "type": "object",
+                "properties": {
+                    "cats": {
+                        "type": "array",
+                        "items": {"$ref": "#/$defs/Category"},
+                    },
+                },
+                "$defs": {
+                    "Category": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "parent": {
+                                "anyOf": [
+                                    {"$ref": "#/$defs/Category"},
+                                    {"type": "null"},
+                                ]
+                            },
+                            "children": {
+                                "type": "array",
+                                "items": {"$ref": "#/$defs/Category"},
+                            },
+                        },
+                    },
+                },
+            }
+        ),
+        output_schema=SchemaRef(type="object"),
+        start="cats",
+        nodes=[
+            ForeachNode.model_validate(
+                {"id": "cats", "type": "foreach", "over": "state.cats", "as": "cat"}
+            ),
+            NodeUse(id="body", type="node", node="noop"),
+        ],
+        edges=[
+            Edge.model_validate({"from": "cats", "outcome": "loop", "to": "body"}),
+            Edge.model_validate({"from": "body", "outcome": "ok", "to": "cats"}),
+            Edge.model_validate({"from": "cats", "outcome": "done", "to": END}),
+        ],
+    )
+
+
+def test_recursive_ref_inventory_stays_bounded() -> None:
+    from wf_api.authoring_contracts import context_path_options_for_node
+
+    options = context_path_options_for_node(_recursive_inventory_workflow(), "body")
+    paths = {option["path"] for option in options}
+    assert "context.foreach.cats.item.parent.name" in paths
+    assert "context.foreach.cats.item.children" in paths
+    # The repeated reference stays selectable but is not expanded again.
+    assert "context.foreach.cats.item.parent.parent" in paths
+    assert "context.foreach.cats.item.parent.parent.name" not in paths