@@ -307,7 +307,24 @@ def _validate_steps(
             wait_for = step_config.get("wait_for")
             if isinstance(wait_for, list):
                 for wid in wait_for:
-                    if isinstance(wid, str) and wid not in seen_ids:
+                    if not isinstance(wid, str):
+                        # A non-string entry (e.g. YAML `wait_for: [123]`) can
+                        # never match a real step id, so the join is silently
+                        # empty at runtime — surface it as a wiring error.
+                        errors.append(
+                            f"Fan-in step {step_id!r}: 'wait_for' entries must "
+                            f"be step-id strings, got {type(wid).__name__} "
+                            f"({wid!r})."
+                        )
+                    elif wid == step_id:
+                        # The fan-in's own id is already in seen_ids by now, so
+                        # a self-reference would pass the membership check below
+                        # while still producing an empty join at runtime.
+                        errors.append(
+                            f"Fan-in step {step_id!r}: 'wait_for' references "
+                            f"itself; a fan-in cannot wait for its own results."
+                        )
+                    elif wid not in seen_ids:
                         errors.append(
                             f"Fan-in step {step_id!r}: 'wait_for' references "
                             f"unknown or not-yet-declared step id {wid!r}."
 
@@ -2057,6 +2057,41 @@ def test_forward_reference_is_rejected(self):
             "unknown or not-yet-declared step id 'later'" in e for e in errors
         )
 
+    def test_self_reference_is_rejected(self):
+        # A fan-in's own id is in scope by the time it is validated, so a
+        # self-reference slips past the membership check while still producing
+        # an empty join at runtime.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+steps:
+  - id: collect
+    type: fan-in
+    wait_for: [collect]
+""")
+        assert any(
+            "references itself" in e and "collect" in e for e in errors
+        )
+
+    def test_non_string_wait_for_entry_is_rejected(self):
+        # A non-string entry (e.g. YAML `wait_for: [123]`) can never match a
+        # real step id, so it must be flagged rather than silently ignored.
+        errors = self._errors("""
+workflow:
+  id: wf
+  name: wf
+  version: "1.0.0"
+steps:
+  - id: collect
+    type: fan-in
+    wait_for: [123]
+""")
+        assert any(
+            "must be step-id strings" in e and "int" in e for e in errors
+        )
+
 
 # ===== Workflow Definition Tests =====