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

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/fluksio/flow/runs.py

@@ -56,7 +56,7 @@ from fluksio.flow.pipeline import CacheHit, NodeOutcome, Pipeline
 from fluksio.flow.queue import WorkItem, WorkQueue
 from fluksio.flow.schemas import FlowDef, NodeDef
 from fluksio.flow.state import MemoryState, StateBackend
-from fluksio.flow.store import FlowStore
+from fluksio.flow.store import FlowNotFound, FlowStore
 from fluksio.models import Run, RunArtifact, RunMetric, RunNode
 from fluksio.sdk.imports import digest_of as files_digest
 
@@ -375,16 +375,20 @@ def _references_in(value: Any, found: set[str], depth: int = 0) -> None:
 
 
 def sweep_artifacts(
-    store: ArtifactStore, state: StateBackend, grace_s: float = 0.0
+    store: ArtifactStore,
+    state: StateBackend,
+    grace_s: float = 0.0,
+    flows: FlowStore | None = None,
 ) -> int:
     """Remove artifact bytes nothing refers to any more. Returns how many went.
 
-    Two things refer to an artifact: a run that recorded it, and a message
-    currently holding it. The second is what makes a stream of media
-    affordable — a camera publishing a frame a second replaces the reference
-    each time, so yesterday's frames are unreferenced by definition and the
-    store does not grow without bound. What a node *returns* is recorded
-    against its run and kept; what it emits along the way is not.
+    Three things refer to an artifact: a run that recorded it, a message
+    currently holding it, and a flow input's ``initial``, published or draft,
+    which a flow that has never started holds nowhere else. The second is what
+    makes a stream of media affordable — a camera publishing a frame a second
+    replaces the reference each time, so yesterday's frames are unreferenced by
+    definition and the store does not grow without bound. What a node *returns*
+    is recorded against its run and kept, what it emits along the way is not.
 
     Skipped entirely while anything is running: a node that stores a checkpoint
     an hour before it returns has neither a row nor a message naming it yet,
@@ -399,6 +403,15 @@ def sweep_artifacts(
             return 0
         keep = set(session.exec(select(col(RunArtifact.digest)).distinct()).all())
 
+    for name in flows.list_flows() if flows else ():
+        for draft in (False, True):
+            try:
+                flow = flows.read_flow(name, draft=draft)
+            except FlowNotFound:
+                continue
+            for declared in flow.inputs:
+                _references_in(declared.initial, keep)
+
     keys = state.keys()
     for start in range(0, len(keys), _SWEEP_BATCH):
         for value in state.get_multi(keys[start : start + _SWEEP_BATCH]).values():

backend/fluksio/main.py

@@ -145,6 +145,7 @@ async def _sweep_artifacts(store: ArtifactStore, controller: FlowController) ->
                 store,
                 controller.state,
                 settings.ARTIFACT_GC_GRACE_S,
+                controller.store,
             )
         except Exception:
             logger.exception("Artifact sweep failed")

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)