From d8a0eef78001f5bb2ad32117c0c68dfb541e37a8 Mon Sep 17 00:00:00 2001 From: lda Date: Sat, 5 Sep 2026 05:00:11 +0700 Subject: [PATCH] fix: delegate structured schema references --- pyproject.toml | 1 + src/wf_api/authoring_contracts.py | 136 +++------ src/wf_core/analysis/__init__.py | 6 - src/wf_core/analysis/context_scopes.py | 342 +---------------------- src/wf_core/conditions.py | 8 +- src/wf_core/schema_navigation.py | 267 ++++++++++++++++++ src/wf_core/validation/context_paths.py | 78 +----- tests/core/test_context_scopes.py | 120 +++++++- tests/wf_api/test_authoring_contracts.py | 7 +- uv.lock | 2 + 10 files changed, 443 insertions(+), 524 deletions(-) create mode 100644 src/wf_core/schema_navigation.py diff --git a/pyproject.toml b/pyproject.toml index 684d7ccc..9e05e887 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py index cd11b064..30142f1c 100644 --- a/src/wf_api/authoring_contracts.py +++ b/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`` - 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. + The shared navigator owns reference and composition semantics. A repeated + reference stays selectable at its own path but is not expanded again, + mirroring the input/state inventory. """ - from .models.authoring_contracts import AuthoringPathOptionPayload as _Payload - - if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH: - return [] - 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 + return _nested_item_view_subpaths( + SchemaNavigator(item_schema).root, + owner_id, + availability, + depth=depth, + prefix_parts=prefix_parts, + active_references=frozenset(), + ) -def _branch_item_children( - branch: Mapping[str, Any], +def _nested_item_view_subpaths( + view: SchemaView, owner_id: str, availability: str, *, depth: int, prefix_parts: tuple[str, ...], - definitions: Mapping[str, Any], - active_refs: frozenset[str], + active_references: frozenset[int], ) -> 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): - 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): + if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH: return [] options: list[_Payload] = [] - for name, sub_schema in properties.items(): - if not isinstance(name, str) or not isinstance(sub_schema, Mapping): - continue - 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 + for child in view.object_children(active_references=active_references): + name = child.name + child_schema = child.view.standalone_schema() 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( - _nested_item_subpaths( - sub_schema, - owner_id, - availability, - depth=depth + 1, - prefix_parts=(*prefix_parts, name), - definitions=definitions, - active_refs=active_refs, + if not child.view.is_array(): + options.extend( + _nested_item_view_subpaths( + child.view, + owner_id, + availability, + depth=depth + 1, + prefix_parts=(*prefix_parts, name), + active_references=child.active_references, + ) ) - ) - return options + 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 _context_entries_for_inventory( diff --git a/src/wf_core/analysis/__init__.py b/src/wf_core/analysis/__init__.py index 3002def1..03c82910 100644 --- a/src/wf_core/analysis/__init__.py +++ b/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", ] diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 7858a299..b8f4e86a 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/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: - resolved_items = _resolve_local_reference(document, items) - except ValueError: + source = SchemaNavigator(document).at_path(foreach.over.parts) + if source is None: return {} - 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( - 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 + items = source.array_items() + if items is None: + return {} + return items.standalone_schema() 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 diff --git a/src/wf_core/conditions.py b/src/wf_core/conditions.py index abead07b..0729251f 100644 --- a/src/wf_core/conditions.py +++ b/src/wf_core/conditions.py @@ -37,9 +37,7 @@ def eval_condition( context=context, ) if isinstance(condition, NotCondition): - return not eval_condition( - condition.arg, state, workflow_input, context=context - ) + return not eval_condition(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( - condition.right, state, workflow_input, context=context - ) + right = resolve_operand(condition.right, state, workflow_input, context=context) if condition.op == "eq": return left == right if condition.op == "ne": diff --git a/src/wf_core/schema_navigation.py b/src/wf_core/schema_navigation.py new file mode 100644 index 00000000..78eda857 --- /dev/null +++ b/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()}" diff --git a/src/wf_core/validation/context_paths.py b/src/wf_core/validation/context_paths.py index 8a4d4d7d..7444c95d 100644 --- a/src/wf_core/validation/context_paths.py +++ b/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 - 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. + permissive unconstrained schemas). Reference resolution is delegated to + the shared Draft 2020-12 navigator; unresolvable refs fail closed. + Composition is a union for path availability: a path is readable when + some structural branch declares it. """ - if not 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) + return SchemaNavigator(schema).first_unknown(parts) 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 3af9a9f2..2a1c02d9 100644 --- a/tests/core/test_context_scopes.py +++ b/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" - ] + 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"} + 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") diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py index 68c20484..62158b0a 100644 --- a/tests/wf_api/test_authoring_contracts.py +++ b/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"), ], diff --git a/uv.lock b/uv.lock index 01e953df..b1960a43 100644 --- a/uv.lock +++ b/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" }, ]