Fix workflow authoring contract integration

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",

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(
         [

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,