sudden Code review

src/wf_core/models/steps.py

@@ -213,9 +213,7 @@ class ForeachNode(BaseModel):
     )
     as_: str = Field(alias="as", description="Context key for the current item.")
     mode: Literal["serial", "concurrent"] = "serial"
-    item_error: ForeachItemErrorPolicy = Field(
-        default_factory=ForeachItemErrorPolicy
-    )
+    item_error: ForeachItemErrorPolicy = Field(default_factory=ForeachItemErrorPolicy)
     concurrent: ForeachConcurrentPolicy | None = None
     on_item_error: Literal["fail", "collect", "skip"] | None = Field(
         default=None,
@@ -252,9 +250,7 @@ class ForeachNode(BaseModel):
         if self.mode == "concurrent" and self.concurrent is None:
             raise ValueError("concurrent foreach requires concurrent policy")
         if self.mode == "serial" and self.concurrent is not None:
-            raise ValueError(
-                "concurrent policy is only valid when mode='concurrent'"
-            )
+            raise ValueError("concurrent policy is only valid when mode='concurrent'")
         return self
 
 

src/wf_core/runtime/foreach_state.py

@@ -35,11 +35,10 @@ class ItemErrorRecord:
             raise WorkflowExecutionError(
                 f"malformed foreach item error record missing {exc.args[0]!r}"
             ) from exc
-        if not isinstance(index, int):
+        if not isinstance(index, int) or index < 0:
             raise WorkflowExecutionError("malformed foreach item error index")
         if not all(
-            isinstance(value, str)
-            for value in (frame_id, node_id, error_type, message)
+            isinstance(value, str) for value in (frame_id, node_id, error_type, message)
         ):
             raise WorkflowExecutionError("malformed foreach item error text fields")
         return cls(
@@ -76,11 +75,16 @@ class PendingItemResult:
     def from_metadata(cls, raw: object) -> PendingItemResult:
         if not isinstance(raw, dict):
             raise WorkflowExecutionError("malformed pending foreach result")
-        index = raw.get("index")
-        frame_id = raw.get("frame_id")
-        status = raw.get("status")
+        try:
+            index = raw["index"]
+            frame_id = raw["frame_id"]
+            status = raw["status"]
+        except KeyError as exc:
+            raise WorkflowExecutionError(
+                f"malformed pending foreach result missing {exc.args[0]!r}"
+            ) from exc
         patch_changes = raw.get("patch_changes", {})
-        if not isinstance(index, int):
+        if not isinstance(index, int) or index < 0:
             raise WorkflowExecutionError("malformed pending foreach result index")
         if not isinstance(frame_id, str):
             raise WorkflowExecutionError("malformed pending foreach result frame id")
@@ -93,10 +97,12 @@ class PendingItemResult:
             index=index,
             frame_id=frame_id,
             status=status,
-            patch=StatePatch(changes=dict(patch_changes)),
-            error=ItemErrorRecord.from_metadata(raw_error)
-            if raw_error is not None
-            else None,
+            patch=StatePatch(changes=patch_changes),
+            error=(
+                ItemErrorRecord.from_metadata(raw_error)
+                if raw_error is not None
+                else None
+            ),
         )
 
     def to_metadata(self) -> dict[str, Any]:
@@ -175,12 +181,17 @@ class ForeachBarrierState:
 
     def save_to_frame(self, frame: ExecutionFrame, foreach_node_id: str) -> None:
         """Store this barrier state in frame metadata under its foreach node id."""
-        raw = frame.metadata.setdefault(_BARRIER_METADATA_KEY, {})
-        if not isinstance(raw, dict):
+        existing = frame.metadata.get(_BARRIER_METADATA_KEY)
+        if existing is None:
+            frame.metadata[_BARRIER_METADATA_KEY] = {
+                foreach_node_id: self.to_metadata()
+            }
+            return
+        if not isinstance(existing, dict):
             raise WorkflowExecutionError(
                 f"malformed foreach barrier table for frame {frame.id!r}"
             )
-        raw[foreach_node_id] = self.to_metadata()
+        existing[foreach_node_id] = self.to_metadata()
 
     def to_metadata(self) -> dict[str, Any]:
         return {

src/wf_core/runtime/scheduler.py

@@ -136,6 +136,10 @@ def block_frame_on_children(
     run: RunState, frame_id: str, child_frame_ids: Sequence[str]
 ) -> None:
     """Mark a frame blocked on child completion and remove it from readiness."""
+    if not child_frame_ids:
+        raise WorkflowExecutionError(
+            f"cannot block frame {frame_id!r} on an empty child set"
+        )
     frame = _frame(run, frame_id)
     run.ready_frame_ids = [item for item in run.ready_frame_ids if item != frame_id]
     frame.status = FrameStatus.BLOCKED

tests/core/test_foreach_barrier_state.py

@@ -64,3 +64,43 @@ def test_foreach_barrier_state_rejects_malformed_metadata() -> None:
 
     with pytest.raises(WorkflowExecutionError, match="next_index"):
         ForeachBarrierState.from_frame(frame, "each")
+
+
+def test_item_error_record_rejects_negative_index() -> None:
+    with pytest.raises(WorkflowExecutionError, match="index"):
+        ItemErrorRecord.from_metadata(
+            {
+                "index": -1,
+                "frame_id": "child",
+                "node_id": "work",
+                "error_type": "ValueError",
+                "message": "bad",
+            }
+        )
+
+
+def test_pending_item_result_reports_missing_required_field() -> None:
+    with pytest.raises(WorkflowExecutionError, match="missing 'frame_id'"):
+        PendingItemResult.from_metadata({"index": 0, "status": "succeeded"})
+
+
+def test_pending_item_result_rejects_negative_index() -> None:
+    with pytest.raises(WorkflowExecutionError, match="index"):
+        PendingItemResult.from_metadata(
+            {"index": -1, "frame_id": "child", "status": "succeeded"}
+        )
+
+
+def test_save_to_frame_rejects_corrupt_table_without_mutating() -> None:
+    frame = ExecutionFrame(
+        id="root",
+        kind="root",
+        node_id="each",
+        metadata={"foreach_barriers": "corrupt"},
+    )
+    barrier = ForeachBarrierState(next_index=1)
+
+    with pytest.raises(WorkflowExecutionError, match="barrier table"):
+        barrier.save_to_frame(frame, "each")
+
+    assert frame.metadata["foreach_barriers"] == "corrupt"

tests/core/test_scheduler.py

@@ -96,6 +96,14 @@ def test_blocked_frame_is_not_selectable_until_woken() -> None:
     assert selected.id == "parent"
 
 
+def test_block_frame_on_children_rejects_empty_child_set() -> None:
+    run = _run()
+    add_frame(run, ExecutionFrame(id="parent", kind="root", node_id="foreach"))
+
+    with pytest.raises(WorkflowExecutionError, match="empty child set"):
+        block_frame_on_children(run, "parent", ())
+
+
 def test_create_run_state_queues_root_frame() -> None:
     workflow = Workflow(
         name="demo",