Artifact sweep keeps what a flow input's initial references, published or draft

backend/tests/test_artifacts.py

@@ -9,8 +9,11 @@ from sqlmodel import Session, col, select
 from fluksio.core.config import settings
 from fluksio.core.db import engine as db_engine
 from fluksio.flow.artifacts import ArtifactStore, VolatileStore
+from fluksio.flow.messages import MessageSpec
 from fluksio.flow.runs import new_run_id, sweep_artifacts
+from fluksio.flow.schemas import FlowDef, FlowInput
 from fluksio.flow.state import MemoryState
+from fluksio.flow.store import FlowStore
 from fluksio.models import Run, RunArtifact
 
 
@@ -112,6 +115,27 @@ def test_a_streamed_chunk_falls_out_once_the_message_moves_on(store):
     assert store.path(second["digest"]) is not None
 
 
+def test_a_sweep_keeps_what_a_flow_input_starts_with(store, tmp_path):
+    """A flow that has never run holds its `initial` in no message yet."""
+    published = store.put([b"published initial"])
+    drafted = store.put([b"draft initial"])
+    orphan = store.put([b"referred to by nothing"])
+
+    flows = FlowStore(tmp_path / "flows")
+    clip = MessageSpec(name="clip", dtype="artifact")
+    flows.write_flow(
+        FlowDef(name="f", inputs=[FlowInput(spec=clip, initial=published)])
+    )
+    flows.write_draft(
+        FlowDef(name="g", inputs=[FlowInput(spec=clip, initial=drafted)]), 0
+    )
+
+    assert sweep_artifacts(store, MemoryState(), grace_s=0, flows=flows) == 1
+    assert store.path(orphan["digest"]) is None
+    for kept in (published, drafted):
+        assert store.path(kept["digest"]) is not None
+
+
 def test_a_ring_drops_the_oldest_once_it_is_full(tmp_path):
     """What makes a camera affordable: the room is fixed, not the history."""
     ring = VolatileStore(tmp_path / "ring", limit_bytes=3000)