diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py index 0cbb8ff0..1c04327e 100644 --- a/src/wf_api/authoring_contracts.py +++ b/src/wf_api/authoring_contracts.py @@ -54,15 +54,21 @@ def schema_path_options( normalized_uses = list(uses) options: list[AuthoringPathOptionPayload] = [] - _append_schema_options( - schema, - location=(), - prefix=prefix, - origin=origin, - uses=normalized_uses, - options=options, - active_references=frozenset(), - ) + try: + _append_schema_options( + schema, + location=(), + prefix=prefix, + origin=origin, + uses=normalized_uses, + options=options, + active_references=frozenset(), + depth=0, + ) + except RecursionError as exc: + raise ValueError( + "schema nesting exceeds the authoring traversal limit" + ) from exc return options @@ -100,7 +106,7 @@ def project_authoring_contract_inventory( state_sources = schema_path_options( state_schema, root="state", - uses=["step_input", "step_output_source", "workflow_output"], + uses=["step_input", "workflow_output"], ) state_targets = schema_path_options( state_schema, @@ -153,7 +159,7 @@ def project_authoring_step_contract( "output_sources": schema_path_options( output_schema, root="step_output", - uses=["step_output_source", "workflow_output"], + uses=["step_output_source"], ), "outcomes": list(outcomes), } @@ -241,7 +247,10 @@ def _append_schema_options( uses: list[AuthoringPathUse], options: list[AuthoringPathOptionPayload], active_references: frozenset[str], + depth: int, ) -> None: + if depth >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH: + return fragment = schema_fragment_at_location(schema, location) resolved = _resolve_local_reference(schema, fragment) properties = resolved.get("properties") @@ -296,6 +305,7 @@ def _append_schema_options( uses=uses, options=options, active_references=next_active_references, + depth=depth + 1, ) diff --git a/src/wf_api/service.py b/src/wf_api/service.py index dacefaf0..bfc958f9 100644 --- a/src/wf_api/service.py +++ b/src/wf_api/service.py @@ -424,12 +424,24 @@ class WorkflowApi: description = raw_step.get("desc") if not isinstance(description, str): description = resolved_contract.description + input_schema = _authoring_schema( + resolved_contract.input_schema, + field_name=f"step {raw_step_id!r} input schema", + root="step_input", + warnings=warnings, + ) + output_schema = _authoring_schema( + resolved_contract.output_schema, + field_name=f"step {raw_step_id!r} output schema", + root="step_output", + warnings=warnings, + ) projected_contract = project_authoring_step_contract( step_id=raw_step_id, label=_step_label(raw_step_id), description=description, - input_schema=resolved_contract.input_schema, - output_schema=resolved_contract.output_schema, + input_schema=input_schema, + output_schema=output_schema, outcomes=resolved_contract.outcomes, ) entry_steps.append(projected_contract) diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 3dd3fc59..c75f9f0e 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -136,6 +136,7 @@ def _analyze(workflow: Workflow) -> _ContextAnalysis: foreach_nodes, node_id, scopes, + scopes_by_node, ) return _ContextAnalysis(fields_by_node, tuple(warnings.values)) @@ -145,6 +146,7 @@ def _available_fields( foreach_nodes: Mapping[str, ForeachNode], node_id: str, scopes: set[FrameScope], + scopes_by_node: Mapping[str, set[FrameScope]], ) -> tuple[ContextFieldAvailability, ...]: del node_id fields_by_name: dict[str, ContextFieldContract] = {} @@ -158,7 +160,13 @@ def _available_fields( *contracts, *foreach_context_fields( foreach.as_, - _foreach_item_schema(workflow, foreach, scope, foreach_nodes), + _foreach_item_schema( + workflow, + foreach, + scopes_by_node.get(foreach.id, {None}), + foreach_nodes, + scopes_by_node, + ), ), ) for contract in contracts: @@ -195,12 +203,26 @@ def _available_fields( def _foreach_item_schema( workflow: Workflow, foreach: ForeachNode, - active_scope: FrameScope, + source_scopes: set[FrameScope], foreach_nodes: Mapping[str, ForeachNode], + scopes_by_node: Mapping[str, set[FrameScope]], ) -> ContextSchema: - source_schema = _schema_at_path( - workflow, foreach.over.root, foreach.over.parts, active_scope, foreach_nodes - ) + source_schemas = [ + _schema_at_path( + workflow, + foreach.over.root, + foreach.over.parts, + source_scope, + foreach_nodes, + scopes_by_node, + ) + for source_scope in sorted(source_scopes, key=lambda value: value or "") + ] + if not source_schemas or any( + schema != source_schemas[0] for schema in source_schemas + ): + return {} + source_schema = source_schemas[0] if not isinstance(source_schema, Mapping): return {} source_type = source_schema.get("type") @@ -212,7 +234,13 @@ def _foreach_item_schema( return {} try: resolved_items = _resolve_local_reference( - _schema_document(workflow, foreach.over.root), items + _schema_document( + workflow, + foreach.over.root, + foreach_nodes=foreach_nodes, + scopes_by_node=scopes_by_node, + ), + items, ) except ValueError: return {} @@ -225,6 +253,7 @@ def _schema_at_path( parts: tuple[str, ...], active_scope: FrameScope, foreach_nodes: Mapping[str, ForeachNode], + scopes_by_node: Mapping[str, set[FrameScope]], ) -> Mapping[str, object] | None: try: schema_document = _schema_document( @@ -232,6 +261,7 @@ def _schema_at_path( root, active_scope=active_scope, foreach_nodes=foreach_nodes, + scopes_by_node=scopes_by_node, ) current: object = schema_document for part in parts: @@ -255,6 +285,7 @@ def _schema_document( *, active_scope: FrameScope = None, foreach_nodes: Mapping[str, ForeachNode] | None = None, + scopes_by_node: Mapping[str, set[FrameScope]] | None = None, ) -> Mapping[str, object]: if root == "input": return workflow.input_schema.model_dump(mode="json", exclude_none=True) @@ -264,7 +295,11 @@ def _schema_document( current: dict[str, object] = { field.name: field.schema for field in STANDARD_CONTEXT_FIELDS } - if active_scope is not None and foreach_nodes is not None: + if ( + active_scope is not None + and foreach_nodes is not None + and scopes_by_node is not None + ): foreach = foreach_nodes.get(active_scope) if foreach is not None: current.update( @@ -275,8 +310,9 @@ def _schema_document( _foreach_item_schema( workflow, foreach, - None, + scopes_by_node.get(foreach.id, {None}), foreach_nodes, + scopes_by_node, ), ) } diff --git a/tests/core/test_context_scopes.py b/tests/core/test_context_scopes.py index 8285ebe2..ba8bff20 100644 --- a/tests/core/test_context_scopes.py +++ b/tests/core/test_context_scopes.py @@ -258,6 +258,49 @@ def test_nested_foreach_replaces_inner_scope_and_restores_outer_scope() -> None: assert "inner_item" not in after_inner +def test_nested_foreach_preserves_context_backed_item_schema() -> None: + workflow = _workflow( + start="outer", + nodes=[ + _foreach("outer", alias="outer_item"), + _foreach("inner", alias="inner_item", over="context.outer_item"), + _node("inner_body"), + ], + edges=[ + {"from": "outer", "outcome": "loop", "to": "inner"}, + {"from": "inner", "outcome": "loop", "to": "inner_body"}, + {"from": "inner", "outcome": "done", "to": END}, + {"from": "inner_body", "outcome": "ok", "to": END}, + {"from": "outer", "outcome": "done", "to": END}, + ], + state_schema={ + "type": "object", + "properties": { + "items": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "object", + "properties": {"id": {"type": "string"}}, + "required": ["id"], + }, + }, + } + }, + }, + ) + + fields = _field_map(workflow, "inner_body") + expected = { + "type": "object", + "properties": {"id": {"type": "string"}}, + "required": ["id"], + } + assert fields["loop_item"].contract.schema == expected + assert fields["inner_item"].contract.schema == expected + + def test_malformed_routes_warn_without_granting_a_scoped_alias() -> None: workflow = _workflow( start="each", diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py index 31056de7..2bb34aae 100644 --- a/tests/wf_api/test_authoring_contracts.py +++ b/tests/wf_api/test_authoring_contracts.py @@ -3,6 +3,7 @@ from __future__ import annotations from wf_api.authoring_contracts import ( context_path_options, project_authoring_contract_inventory, + project_authoring_step_contract, schema_path_options, ) @@ -217,6 +218,19 @@ def test_schema_path_options_stops_expanding_recursive_local_definition() -> Non ] +def test_schema_path_options_bounds_deep_inline_objects() -> None: + schema: dict[str, object] = {"type": "object", "properties": {}} + current = schema + for index in range(60): + child: dict[str, object] = {"type": "object", "properties": {}} + current["properties"] = {f"level_{index}": child} + current = child + + options = schema_path_options(schema, root="input", uses=["step_input"]) + + assert len(options) == 32 + + def test_schema_path_options_returns_empty_schema_for_unconstrained_property() -> None: options = schema_path_options( { @@ -319,7 +333,7 @@ def test_project_authoring_contract_inventory_composes_pure_inputs() -> None: "schema": {"type": "string"}, "required": False, "availability": "available", - "uses": ["step_input", "step_output_source", "workflow_output"], + "uses": ["step_input", "workflow_output"], }, ] assert inventory["step_input_targets"] == [step_input_target] @@ -331,6 +345,37 @@ def test_project_authoring_contract_inventory_composes_pure_inputs() -> None: assert inventory["warnings"] == ["selected step has conditional context"] +def test_authoring_paths_exclude_incompatible_binding_roles() -> None: + inventory = project_authoring_contract_inventory( + workspace_id="workspace-1", + revision=1, + selected_step_id=None, + input_schema={"type": "object"}, + state_schema={ + "type": "object", + "properties": {"answer": {"type": "string"}}, + }, + output_schema={"type": "object"}, + ) + step_contract = project_authoring_step_contract( + step_id="fetch", + label="Fetch", + description=None, + input_schema={"type": "object"}, + output_schema={ + "type": "object", + "properties": {"answer": {"type": "string"}}, + }, + outcomes=["ok"], + ) + + state_source = inventory["readable_sources"][0] + assert state_source["path"] == "state.answer" + assert "step_output_source" not in state_source["uses"] + assert step_contract["output_sources"][0]["path"] == "step_output.answer" + assert "workflow_output" not in step_contract["output_sources"][0]["uses"] + + def test_context_path_options_are_step_input_only() -> None: options = context_path_options( [ diff --git a/tests/wf_api/test_drafts_service.py b/tests/wf_api/test_drafts_service.py index 72ee166e..5c56fc80 100644 --- a/tests/wf_api/test_drafts_service.py +++ b/tests/wf_api/test_drafts_service.py @@ -277,6 +277,56 @@ async def test_inspect_draft_authoring_contract_preserves_empty_capability_schem assert inventory["step_output_sources"] == [] +@pytest.mark.asyncio +@pytest.mark.parametrize( + "invalid_property_schema", + [ + {"$ref": "https://example.com/request.json"}, + {"$ref": 7}, + ], +) +async def test_inspect_draft_authoring_contract_warns_for_invalid_capability_schema( + tmp_path: Path, + invalid_property_schema: dict[str, object], +) -> None: + draft_api, service, authoring = _draft_api( + FileWorkflowArtifactStore(tmp_path / "authoring_contract_bad_capability") + ) + service.register_connection( + ConnectionConfig(id="demo.personal", server="demo", account="personal") + ) + service.register_specs( + "demo.personal", + replace( + echo_tool, + input_schema_contract={ + "type": "object", + "properties": {"text": invalid_property_schema}, + }, + ), + ) + await draft_api.create_draft_workspace( + workspace_id="authoring", + draft=_echo_draft(), + ) + api = WorkflowApi(authoring.context) + + inventory = await api.inspect_draft_authoring_contract( + workspace_id="authoring", + revision=1, + selected_step_id="echo", + ) + + assert inventory["entry_steps"][0]["input_targets"] == [] + assert inventory["entry_steps"][0]["output_sources"] + assert inventory["step_input_targets"] == [] + assert inventory["step_output_sources"] + assert any( + "echo" in warning and "input schema" in warning + for warning in inventory["warnings"] + ) + + @pytest.mark.asyncio async def test_inspect_draft_authoring_contract_rejects_unknown_selected_step( tmp_path: Path,