From 3016fd263bb18b30bbbd59417078832ce23beafb Mon Sep 17 00:00:00 2001 From: lda Date: Sat, 5 Sep 2026 00:10:09 +0700 Subject: [PATCH] fix: inline nested schema refs in structured foreach context --- src/wf_core/analysis/context_scopes.py | 77 ++++++++++++++++--- .../test_structured_context_validation.py | 59 ++++++++++++++ 2 files changed, 125 insertions(+), 11 deletions(-) diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 39c12e39..e8e05872 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -218,6 +218,9 @@ def _available_fields( """ contracts = list(STANDARD_CONTEXT_FIELDS) if not stack: + # The empty foreach map stays readable at root, matching + # ``root_context_schema`` and the validation pass. + contracts.append(structured_foreach_contract({})) return tuple( ContextFieldAvailability( contract=ContextFieldContract( @@ -371,20 +374,18 @@ def _foreach_item_schema( items = source_schema.get("items") if not is_array or not isinstance(items, Mapping): return {} + document = _schema_document( + workflow, + foreach.over.root, + stack=controller_stack, + foreach_nodes=foreach_nodes, + owner_stack_by_node=owner_stack_by_node, + ) try: - resolved_items = _resolve_local_reference( - _schema_document( - workflow, - foreach.over.root, - stack=controller_stack, - foreach_nodes=foreach_nodes, - owner_stack_by_node=owner_stack_by_node, - ), - items, - ) + resolved_items = _resolve_local_reference(document, items) except ValueError: return {} - return deepcopy(dict(resolved_items)) + return deepcopy(dict(_inline_local_refs(document, resolved_items))) def _schema_at_path( @@ -477,6 +478,60 @@ def _schema_document( return {} +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 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. + """ + + 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): + if reference in active: + return node + try: + resolved = _resolve_local_reference(root_schema, node) + except ValueError: + return node + return inline(resolved, active | {reference}, 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) + return inlined + + inlined = inline(candidate, frozenset(), 0) + if not isinstance(inlined, Mapping): + return candidate + return inlined + + def _resolve_local_reference( root_schema: Mapping[str, object], candidate: Mapping[str, object], diff --git a/tests/core/test_structured_context_validation.py b/tests/core/test_structured_context_validation.py index 2eaebb2c..98107e47 100644 --- a/tests/core/test_structured_context_validation.py +++ b/tests/core/test_structured_context_validation.py @@ -533,3 +533,62 @@ def test_object_expression_and_nested_conditions_report_exact_paths() -> None: ) is not None ) + + +def _nested_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/Order"}, + }, + }, + "$defs": { + "Order": { + "type": "object", + "properties": { + "sku": {"type": "string"}, + "detail": {"$ref": "#/$defs/Detail"}, + }, + "required": ["sku", "detail"], + }, + "Detail": { + "type": "object", + "properties": {"name": {"type": "string"}}, + "required": ["name"], + }, + }, + } + ) + return workflow + + +def test_nested_ref_item_subpath_is_accepted() -> None: + from wf_core.validation import validate_workflow + + workflow = _nested_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_nested_ref_item_unknown_leaf_reports_available_keys() -> None: + from wf_core.validation import validate_workflow + + workflow = _nested_ref_workflow( + work_path="context.foreach.orders.item.detail.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: name" in issue.message