fix(storage): reject flushing lazy source step

dataflow/utils/storage.py

@@ -352,6 +352,16 @@ def clean_surrogates(obj):
 
     # ---------- 落盘 ----------
     def flush_step(self, step: int):
+        if step <= 0:
+            raise ValueError(
+                f"Cannot flush step {step}. LazyFileStorage reserves step 0 for "
+                "the original input source, and negative steps do not identify "
+                "writable outputs. Flushing step 0 could overwrite or reformat "
+                "the user's source file. Only output steps greater than 0 can be "
+                "flushed; pass a positive output step containing data buffered "
+                "by write()."
+            )
+
         with self._lock:
             if step not in self._buffers:
                 self.logger.info(f"No buffer for step {step}; nothing to flush.")

test/cpu_only/test_lazy_file_storage_flush.py

@@ -132,6 +132,20 @@ def test_explicit_flush_step_can_persist_an_older_output(source_file):
     ]
 
 
+def test_flush_step_rejects_source_step(source_file):
+    original = source_file.read_bytes()
+    storage = make_storage(source_file)
+    assert storage.step().read("dict") == [{"id": 1, "text": "original, text"}]
+
+    with pytest.raises(
+        ValueError,
+        match=r"Cannot flush step 0.*original input source.*greater than 0",
+    ):
+        storage.flush_step(0)
+
+    assert source_file.read_bytes() == original
+
+
 @pytest.mark.parametrize("flush_all_steps", [False, True])
 def test_latest_output_can_be_updated_after_flushing(source_file, flush_all_steps):
     original = source_file.read_bytes()