Say what a node returned when it is not a dict of ports

Outputs are keyed by port, so a bare value cannot be one. The single mapping
point every caller routes through raised a bare AttributeError from
retval.items(); it now names the problem, and the message reaches the node the
way its other errors do.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01H7LwYgJfpkbLCTeiAf8U4A

backend/tests/flow/test_pipeline.py

@@ -1,5 +1,6 @@
 """The wiring fundamentals: name binding, fan-in, namespaces, validation."""
 
+from app.flow.events import EventBus
 from app.flow.messages import DType, MessageSpec
 from app.flow.nodes import Node
 from app.flow.pipeline import Pipeline
@@ -150,6 +151,23 @@ def test_a_failing_node_does_not_stop_its_siblings():
     assert pipeline.state["f.ok"] == 1.0
 
 
+def test_a_node_returning_something_other_than_a_dict_says_what_is_wrong():
+    """Outputs are keyed by port, so a bare value cannot be one of them."""
+    events = []
+
+    def wrong(params):
+        return 42.0
+
+    bus = EventBus()
+    bus.publish = events.append  # type: ignore[method-assign]
+    node = make_node("n", "f", wrong, provides=[spec("out")])
+    Pipeline(nodes=[node], events=bus).run()
+
+    (error,) = [e for e in events if e["type"] == "node_error"]
+    assert "NodeOutputError" in error["error"]
+    assert "returned float" in error["error"]
+
+
 def test_values_carry_timestamps():
     node = make_node("n", "f", lambda params: {"out": 1.0}, provides=[spec("out")])
     pipeline = Pipeline(nodes=[node])