Keep the file name a node gave an artifact

A run_artifact row is keyed by the message the bytes left on, and that was
also the only name it could answer with — so an `@run:` reference resolved
through the row was the same bytes under a name its producer never chose.
The row now records the file name beside the message name; rows written
before the column answer as they always did.

The fallback also checks the bytes are still in the store, which the bare
digest spelling beside it has always done. A missing blob now fails at
submit rather than in the middle of the run that wanted it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/fluksio/flow/runs.py

@@ -184,13 +184,17 @@ def resolve_references(
     with Session(db_engine) as session:
         for key, text in pending.items():
             if text.startswith(RUN_REF_PREFIX):
-                resolved[key] = _from_run(session, key, text[len(RUN_REF_PREFIX) :])
+                resolved[key] = _from_run(
+                    session, key, text[len(RUN_REF_PREFIX) :], artifacts
+                )
             else:
                 resolved[key] = _from_digest(session, key, text, artifacts)
     return resolved
 
 
-def _from_run(session: Session, key: str, spelling: str) -> Any:
+def _from_run(
+    session: Session, key: str, spelling: str, artifacts: ArtifactStore | None = None
+) -> Any:
     """``<run id>.<output>`` as the value that run produced."""
     run_id, _, output = spelling.partition(".")
     if not run_id or not output:
@@ -210,18 +214,25 @@ def _from_run(session: Session, key: str, spelling: str) -> Any:
         return copy.deepcopy(result[output])
 
     # The artifact rows are the fallback: bytes a node made that the flow never
-    # declared as an output. They carry the message name instead, which loads
-    # the same bytes either way.
+    # declared as an output. A row records what the node called the file, so
+    # the reference rebuilt here is the one its producer made — apart from the
+    # rows written before there was a column to keep it in, which answer with
+    # the message name as they always did.
     rows = session.exec(
         select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
     ).all()
     for row in rows:
         if output in (row.name, row.name.rsplit(".", 1)[-1]):
+            if artifacts is not None and artifacts.path(row.digest) is None:
+                raise RunRejected(
+                    f"Parameter '{key}': run '{run_id}' made '{output}', but its "
+                    "bytes are gone from this installation's store"
+                )
             return {
                 "digest": row.digest,
                 "size": row.size,
                 "media_type": row.media_type or "application/octet-stream",
-                "name": row.name,
+                "name": row.filename or row.name,
             }
     known = ", ".join(sorted({*result, *(row.name for row in rows)})) or "none"
     raise RunRejected(
@@ -252,7 +263,7 @@ def _from_digest(
         "digest": row.digest,
         "size": row.size,
         "media_type": row.media_type or "application/octet-stream",
-        "name": row.name,
+        "name": row.filename or row.name,
     }
 
 
@@ -982,6 +993,7 @@ class RunService:
                         RunArtifact(
                             run_id=run_id,
                             name=message[:255],
+                            filename=str(ref.get("name") or "")[:255] or None,
                             node=outcome.node[:255],
                             digest=str(ref.get("digest") or "")[:71],
                             size=int(ref.get("size") or 0),