Guard PWD conversion

notebooks/user-guide.ipynb

@@ -3400,6 +3400,11 @@
     "- Only **flat workflows** (all children are atomic) can be converted.\n",
     "- **Default values** must be supplied for every workflow input (PWD input nodes\n",
     "  carry concrete values).\n",
+    "- Every node must have **exactly one output**. PWD reads multiple output ports as\n",
+    "  keys of a single returned dict, while flowrep reads them as unpacking a tuple,\n",
+    "  so rather than silently mistranslate, the converter raises\n",
+    "  `OutputContractError`. Relatedly, a single output port's *name* is not\n",
+    "  preserved — it becomes PWD's unnamed `sourcePort` and returns as `__result__`.\n",
     "\n",
     "### 7.1 flowrep → PWD"
    ],

src/flowrep/api/tools.py

@@ -6,6 +6,9 @@
 """
 
 from flowrep.compiler.source import flowrep2python as flowrep2python
+from flowrep.converters.python_workflow_definition import (
+    OutputContractError as OutputContractError,
+)
 from flowrep.converters.python_workflow_definition import flowrep2pwd as flowrep2pwd
 from flowrep.converters.python_workflow_definition import pwd2flowrep as pwd2flowrep
 from flowrep.parsers.atomic_parser import atomic as atomic

src/flowrep/converters/python_workflow_definition.py

@@ -4,6 +4,24 @@
 The ``python_workflow_definition`` (pwd) package is an **optional** dependency.
 It represents workflows as flat, non-nested DAGs of atomic function calls with
 explicit input/output nodes carrying JSON-serializable default values.
+
+The two formats hold different axioms about how a function's return value maps
+onto a node's output ports:
+
+===========  ==========================  ================================
+             whole return value          component of return
+===========  ==========================  ================================
+pwd          ``sourcePort: null``        ``sourcePort: "<key>"`` →
+                                         ``result["<key>"]``
+flowrep      exactly one output port     N>1 ports unpack an N-tuple
+===========  ==========================  ================================
+
+They overlap only on *exactly one output, meaning the whole return value*.
+Anything outside that overlap raises :class:`OutputContractError` rather than
+being mistranslated into a recipe that computes the wrong answer.  As a
+corollary, a flowrep node's sole output port name is not preserved across a
+round-trip: it becomes pwd's unnamed ``sourcePort`` and returns as
+:data:`_DEFAULT_OUTPUT_PORT`.
 """
 
 from __future__ import annotations
@@ -36,6 +54,23 @@
 _PORT_SANITIZE_PREFIX: str = "flowrep_sanitized_"
 
 
+class OutputContractError(ValueError):
+    """
+    A node's output contract is not representable in the target format.
+
+    flowrep and pwd disagree about how a function's return value maps onto
+    output ports.  pwd says ``sourcePort: null`` is the entire return value and
+    a named ``sourcePort`` is a key of a returned dict; flowrep says a single
+    output port is the entire return value and N>1 ports unpack an N-tuple.
+    The two overlap only on *exactly one output, meaning the whole return
+    value*.  Converting a node outside that overlap would produce a recipe that
+    executes incorrectly, so we refuse rather than mistranslate.
+
+    Subclasses :class:`ValueError` so that callers already handling the
+    converter's other validation failures keep working.
+    """
+
+
 def _needs_sanitization(port: str) -> bool:
     """Return ``True`` if *port* is not a valid flowrep :class:`Label`."""
     return not base_models.is_valid_label(port)
@@ -55,7 +90,12 @@ def _sanitize_port(port: str) -> str:
 
 
 def _desanitize_port(port: str) -> str:
-    """Reverse :func:`_sanitize_port` when converting back to pwd."""
+    """
+    Reverse :func:`_sanitize_port` when converting back to pwd.
+
+    Only input (``targetPort``) names need this; output ports never round-trip
+    through pwd by name — see :func:`_build_pwd_edges`.
+    """
     if port.startswith(_PORT_SANITIZE_PREFIX):
         candidate = port[len(_PORT_SANITIZE_PREFIX) :]
         if candidate and _needs_sanitization(candidate):
@@ -77,8 +117,13 @@ def pwd2flowrep(
         A ``(WorkflowRecipe, defaults)`` pair where *defaults* maps each
         workflow-input name to the default value carried by the corresponding
         PWD input node.
+
+    Raises:
+        OutputContractError: If any function node is consumed via a named
+            ``sourcePort``.
     """
     input_nodes, output_nodes, function_nodes = _categorize_pwd_nodes(wf.nodes)
+    _validate_pwd_output_contracts(wf.edges, function_nodes)
 
     label_map = _build_label_map(function_nodes)
     node_inputs, node_outputs = _collect_function_node_ports(
@@ -136,8 +181,10 @@ def flowrep2pwd(
     Raises:
         ValueError: If any child is non-atomic, or if *terminal_inputs* does
             not exactly cover the workflow's inputs.
+        OutputContractError: If any child has more than one output port.
     """
     _validate_flat_workflow(wf)
+    _validate_flowrep_output_contract(wf)
     _validate_terminal_inputs(wf, terminal_inputs)
 
     id_counter = _IdCounter()
@@ -216,6 +263,45 @@ def _categorize_pwd_nodes(
     return input_nodes, output_nodes, function_nodes
 
 
+def _validate_pwd_output_contracts(
+    edges: list[pwd.PythonWorkflowDefinitionEdge],
+    function_nodes: dict[int, pwd.PythonWorkflowDefinitionFunctionNode],
+) -> None:
+    """
+    Raise :class:`OutputContractError` for function nodes consumed by key.
+
+    A named ``sourcePort`` tells pwd to subscript the node's single return value
+    with that key.  A flowrep atomic node cannot express that: its outputs are
+    either the whole return value (one port) or a tuple unpacking (many ports).
+    Inspects raw ``sourcePort`` values, before sanitisation, so the message
+    quotes the author's own port names.
+    """
+    named: dict[int, list[str]] = {}
+    for edge in edges:
+        if edge.source not in function_nodes:
+            continue
+        port = edge.sourcePort
+        if port is None or port == pwd.INTERNAL_DEFAULT_HANDLE:
+            continue
+        ports = named.setdefault(edge.source, [])
+        if port not in ports:
+            ports.append(port)
+
+    if named:
+        offenders = [
+            f"node {nid} ({function_nodes[nid].value}) is consumed via "
+            f"sourcePort(s) {sorted(named[nid])}"
+            for nid in sorted(named)
+        ]
+        raise OutputContractError(
+            "pwd2flowrep requires every function node to be consumed via a null "
+            "sourcePort. pwd reads a named sourcePort as a key of a single "
+            "returned dict, which a flowrep atomic node cannot express — its "
+            "outputs are either the whole return value (one port) or a tuple "
+            "unpacking (many). Offending nodes: " + "; ".join(offenders) + "."
+        )
+
+
 def _build_label_map(
     function_nodes: dict[int, pwd.PythonWorkflowDefinitionFunctionNode],
 ) -> dict[int, str]:
@@ -237,22 +323,16 @@ def _build_label_map(
     return label_map
 
 
-def _resolve_source_port(port: str | None) -> str:
-    """Map pwd's internal sourcePort to a flowrep output-port name."""
-    if port is None or port == pwd.INTERNAL_DEFAULT_HANDLE:
-        return _DEFAULT_OUTPUT_PORT
-    return _sanitize_port(port)
-
-
 def _collect_function_node_ports(
     edges: list[pwd.PythonWorkflowDefinitionEdge],
     function_nodes: dict[int, pwd.PythonWorkflowDefinitionFunctionNode],
 ) -> tuple[dict[int, list[str]], dict[int, list[str]]]:
     """
     Collect ordered, unique input/output port names for each function node.
 
-    Port names that are not valid flowrep Labels are sanitized via
-    :func:`_sanitize_port`.
+    Input port names that are not valid flowrep Labels are sanitized via
+    :func:`_sanitize_port`.  Output ports are always the single
+    :data:`_DEFAULT_OUTPUT_PORT`.
     """
     node_inputs: dict[int, list[str]] = {nid: [] for nid in function_nodes}
     node_outputs: dict[int, list[str]] = {nid: [] for nid in function_nodes}
@@ -265,12 +345,13 @@ def _collect_function_node_ports(
             if port not in ports:
                 ports.append(port)
 
-        # Source ports (outputs of the function node)
+        # Outputs of the function node.  Guaranteed null-sourced by
+        # _validate_pwd_output_contracts, so every such node has exactly the
+        # one default-named output port.
         if edge.source in function_nodes:
-            port = _resolve_source_port(edge.sourcePort)
             ports = node_outputs[edge.source]
-            if port not in ports:
-                ports.append(port)
+            if _DEFAULT_OUTPUT_PORT not in ports:
+                ports.append(_DEFAULT_OUTPUT_PORT)
 
     return node_inputs, node_outputs
 
@@ -345,7 +426,7 @@ def _build_flowrep_edges(
             # Function node → workflow output
             out_name = output_nodes[edge.target].name
             source_label = label_map[edge.source]
-            source_port = _resolve_source_port(edge.sourcePort)
+            source_port = _DEFAULT_OUTPUT_PORT
             fr_output_edges[edge_models.OutputTarget(port=out_name)] = (
                 edge_models.SourceHandle(node=source_label, port=source_port)
             )
@@ -354,7 +435,7 @@ def _build_flowrep_edges(
             # Function node → function node (sibling edge)
             source_label = label_map[edge.source]
             target_label = label_map[edge.target]
-            source_port = _resolve_source_port(edge.sourcePort)
+            source_port = _DEFAULT_OUTPUT_PORT
             target_port = _sanitize_port(edge.targetPort)
             fr_edges[edge_models.TargetHandle(node=target_label, port=target_port)] = (
                 edge_models.SourceHandle(node=source_label, port=source_port)
@@ -385,6 +466,33 @@ def _validate_flat_workflow(wf: workflow_recipe.WorkflowRecipe) -> None:
             )
 
 
+def _validate_flowrep_output_contract(wf: workflow_recipe.WorkflowRecipe) -> None:
+    """
+    Raise :class:`OutputContractError` if any child has more than one output.
+
+    Must be called *after* :func:`_validate_flat_workflow`, which guarantees
+    every child is an :class:`AtomicRecipe`.
+    """
+    offenders: list[str] = []
+    for label, node in wf.nodes.items():
+        # Guaranteed by _validate_flat_workflow
+        assert isinstance(node, atomic_recipe.AtomicRecipe)
+        if len(node.outputs) > 1:
+            offenders.append(
+                f"'{label}' ({node.fully_qualified_name}) has "
+                f"{len(node.outputs)} outputs {node.outputs}"
+            )
+    if offenders:
+        raise OutputContractError(
+            "flowrep2pwd requires every child to have exactly one output. "
+            "flowrep reads multiple output ports as unpacking a tuple, whereas "
+            "pwd reads them as keys of a single returned dict, so the converted "
+            "workflow would not execute correctly. Offending nodes: "
+            + "; ".join(offenders)
+            + "."
+        )
+
+
 def _validate_terminal_inputs(
     wf: workflow_recipe.WorkflowRecipe,
     terminal_inputs: dict[str, Any],
@@ -405,19 +513,6 @@ def _validate_terminal_inputs(
         )
 
 
-def _flowrep_port_to_pwd_source_port(port: str) -> str | None:
-    """
-    Map a flowrep output-port name back to a pwd ``sourcePort`` value.
-
-    Returns ``None`` for the default-output sentinel so that the PWD edge
-    validator stores it as :data:`pwd.INTERNAL_DEFAULT_HANDLE`.  Otherwise
-    reverses any sanitisation applied by :func:`_sanitize_port`.
-    """
-    if port == _DEFAULT_OUTPUT_PORT:
-        return None
-    return _desanitize_port(port)
-
-
 def _build_pwd_edges(
     wf: workflow_recipe.WorkflowRecipe,
     input_node_ids: dict[str, int],
@@ -430,6 +525,11 @@ def _build_pwd_edges(
     Edges are emitted in a deterministic order that preserves the input-port
     ordering of each child node — this is important for consumers that rely on
     edge-list order (e.g. ``get_list``).
+
+    Every function-node source emits ``sourcePort=None``.  A flowrep child has
+    exactly one output (enforced by :func:`_validate_flowrep_output_contract`) and that
+    output *is* the whole return value, which is precisely what pwd spells as a
+    null ``sourcePort``.  The flowrep port's name is therefore not carried over.
     """
     pwd_edges: list[pwd.PythonWorkflowDefinitionEdge] = []
 
@@ -456,7 +556,7 @@ def _build_pwd_edges(
                 pwd_edges.append(
                     pwd.PythonWorkflowDefinitionEdge(
                         source=func_node_ids[source.node],
-                        sourcePort=_flowrep_port_to_pwd_source_port(source.port),
+                        sourcePort=None,
                         target=func_node_ids[label],
                         targetPort=target_port_pwd,
                     )
@@ -480,7 +580,7 @@ def _build_pwd_edges(
                 pwd_edges.append(
                     pwd.PythonWorkflowDefinitionEdge(
                         source=func_node_ids[source.node],
-                        sourcePort=_flowrep_port_to_pwd_source_port(source.port),
+                        sourcePort=None,
                         target=output_node_ids[port],
                         targetPort=None,
                     )

tests/flowrep_static/library.py

@@ -216,3 +216,13 @@ def single_autoencoder(only):
     dc = Single(only)
     o = Single.flowrep_recipe_unpacking(dc)
     return o
+
+
+def prod_and_div_dict(x, y):
+    """
+    PWD's output axiom: several outputs are keys of one returned dict.
+
+    Deliberately *not* decorated -- pwd's ``purepython`` runner filters nodes
+    with :func:`inspect.isfunction`.
+    """
+    return {"prod": x * y, "div": x / y}

tests/flowrep_static/python-workflow-definition/mono-workflow.json

@@ -0,0 +1,23 @@
+{
+  "version": "0.1.0",
+  "nodes": [
+    {"id": 0, "type": "function", "value": "workflow.scale"},
+    {"id": 1, "type": "function", "value": "workflow.scale"},
+    {"id": 2, "type": "function", "value": "python_workflow_definition.shared.get_list"},
+    {"id": 3, "type": "function", "value": "workflow.summarize"},
+    {"id": 4, "type": "input", "value": 2, "name": "x"},
+    {"id": 5, "type": "input", "value": {"a": [1, 2], "b": {"c": 3}}, "name": "options"},
+    {"id": 6, "type": "output", "name": "summary"},
+    {"id": 7, "type": "output", "name": "passthrough"}
+  ],
+  "edges": [
+    {"target": 0, "targetPort": "v", "source": 4, "sourcePort": null},
+    {"target": 1, "targetPort": "v", "source": 4, "sourcePort": null},
+    {"target": 2, "targetPort": "0", "source": 0, "sourcePort": null},
+    {"target": 2, "targetPort": "1", "source": 1, "sourcePort": null},
+    {"target": 3, "targetPort": "values", "source": 2, "sourcePort": null},
+    {"target": 3, "targetPort": "options", "source": 5, "sourcePort": null},
+    {"target": 6, "targetPort": null, "source": 3, "sourcePort": null},
+    {"target": 7, "targetPort": null, "source": 5, "sourcePort": null}
+  ]
+}