Artifacts: bytes a node produced, addressed by their content

A checkpoint is not a message. DType.ARTIFACT carries a reference — digest,
size, media type, name — so everything on the wire stays JSON and thirty
megabytes never sit in Redis, which answers the vision's open binary-payload
question by narrowing it: inline codecs would only serve payloads too small to
be worth a round trip, and nothing asks for that.

The store is content-addressed rather than per-run, for three reasons that all
pay later: a sweep whose fifty configs share one preprocessed input stores it
once, a reference stays valid however it is passed around because it names
content instead of a location, and the digest is what a stage cache will
compare — so building it in now is what keeps that from being a change to the
message contract.

Node code calls fluksio.save_artifact/load_artifact and cannot tell whether it
is writing the engine's own directory or putting bytes over HTTP, which is
what will let the same flow run on a remote worker unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD

backend/tests/flow/test_workers.py

@@ -7,6 +7,9 @@ from collections.abc import Iterator
 
 import pytest
 
+from app.flow.artifacts import ArtifactStore
+from app.flow.messages import DType, MessageSpec
+from app.flow.worker_main import ARTIFACT_DIR_ENV
 from app.flow.workers import NodeCancelled, NodeTimeout, PythonWorkerPool
 
 
@@ -280,3 +283,45 @@ def test_cancelling_one_run_leaves_the_same_node_in_another_alone(pool):
     assert pool.cancel("demo.hold", run_id="run-a") is True
     started.wait(timeout=5)
     thread.join(timeout=5)
+
+
+def test_a_node_saves_and_loads_an_artifact(tmp_path):
+    # Bytes never travel as a message: the node stores them and returns a
+    # reference, which the next node opens.
+    store = ArtifactStore(tmp_path / "artifacts")
+    pool = PythonWorkerPool(
+        python=sys.executable, size=1, env={ARTIFACT_DIR_ENV: str(store.root)}
+    )
+    pool.start()
+    try:
+        ref = pool.run(
+            "demo",
+            "save",
+            "import fluksio\n"
+            "def process(params):\n"
+            "    return {'weights': fluksio.save_artifact(b'x' * 2048, 'w.npz')}\n",
+            {},
+            {},
+            "demo.save",
+            timeout=10,
+        )["weights"]
+
+        assert ref["size"] == 2048
+        assert MessageSpec(name="weights", dtype=DType.ARTIFACT).check(ref) is None
+        assert store.path(ref["digest"]) is not None
+
+        loaded = pool.run(
+            "demo",
+            "load",
+            "import fluksio\n"
+            "def process(weights, params):\n"
+            "    with open(fluksio.load_artifact(weights), 'rb') as f:\n"
+            "        return {'size': len(f.read())}\n",
+            {"weights": ref},
+            {},
+            "demo.load",
+            timeout=10,
+        )
+        assert loaded == {"size": 2048}
+    finally:
+        pool.stop()