A cached node keeps its curve, and any input can name a run's output

A cache hit still replays no emissions — those values were the story of an
execution that is not happening — but the run they were recorded in is now
written on the row (`run_node.cached_from`), and the metrics endpoints read the
series back from there. So a reused run answers `run.metrics("train.loss")`
with the same points the run that trained did, rather than looking like a run
that produced no numbers at all. Pointed at rather than copied: a sweep of 500
reusing one frozen node would otherwise duplicate its curve 500 times.

That needed the cross-flow restore fixed first. The cache key has no flow in
it while the stored outputs are named for the flow that produced them, so
`quick.prepare` getting a hit from `train` wrote `train.dataset` into `quick`'s
state and the next node was called without its argument. One rule now covers
both halves: `requalify` reads a name owned by one flow as the same name in
another, applied to the restored outputs, to the node id behind the pointer,
and to the series names on the way out. Reuse across flows is kept.

Also: `@run:<id>.<output>` and a bare `sha256:` digest resolve on every input,
not only artifacts. Chaining a run's json config into the next one from a shell
meant pasting the whole object inline, and the CLI could not even send the
spelling — `_coerce` died in `json.loads` before the engine saw it. Both
spellings are reserved on every input now, `str` included, and `_from_run`
returns whatever the run's result holds rather than only a reference.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dp9L6gakMVro1K2C5zdtBE

backend/fluksio/flow/runs.py

@@ -25,6 +25,7 @@ Three things make a run different from a cascade, and each is deliberate:
 
 from __future__ import annotations
 
+import copy
 import hashlib
 import json
 import logging
@@ -46,8 +47,8 @@ from sqlmodel import Session, col, select
 from fluksio.core.db import engine as db_engine
 from fluksio.flow.artifacts import ArtifactStore, is_reference, valid_digest
 from fluksio.flow.controller import FlowController, RunContext
-from fluksio.flow.messages import DType, qualify
-from fluksio.flow.pipeline import NodeOutcome, Pipeline
+from fluksio.flow.messages import qualify
+from fluksio.flow.pipeline import CacheHit, NodeOutcome, Pipeline
 from fluksio.flow.queue import WorkItem, WorkQueue
 from fluksio.flow.schemas import FlowDef
 from fluksio.flow.state import MemoryState, StateBackend
@@ -152,25 +153,27 @@ RUN_REF_PREFIX = "@run:"
 def resolve_references(
     flow: FlowDef, params: dict[str, Any], artifacts: ArtifactStore | None = None
 ) -> dict[str, Any]:
-    """Turn the text spellings of an artifact input into the reference itself.
+    """Turn the text spelling of a run's output into the value itself.
 
     A python caller hands one run's output straight to the next, because it has
-    the reference in its hand. A shell does not, and pasting the whole object
-    is not a command anyone wants to type — so an artifact input also takes
-    ``@run:<id>.<output>``, naming what a run produced, or a bare
-    ``sha256:...`` digest naming the bytes. Resolved here rather than in each
-    client, so the CLI, the browser and a python caller all mean the same
-    thing by the same string.
+    the value in its hand. A shell does not, and pasting a config object or an
+    artifact reference is not a command anyone wants to type — so any declared
+    input also takes ``@run:<id>.<output>``, naming what a run produced, and a
+    bare ``sha256:...`` digest naming bytes. Resolved here rather than in each
+    client, so the CLI, the browser and a python caller all mean the same thing
+    by the same string.
+
+    Both spellings are therefore reserved on every input, ``str`` included: an
+    input that has to carry one of them literally is asking for a value this
+    engine reads as a name.
     """
-    wanted = {
-        declared.spec.name
-        for declared in flow.inputs
-        if declared.spec.dtype is DType.ARTIFACT
-    }
+    declared = {one.spec.name for one in flow.inputs}
     pending = {
         key: value
         for key, value in params.items()
-        if key in wanted and isinstance(value, str)
+        if key in declared
+        and isinstance(value, str)
+        and (value.startswith(RUN_REF_PREFIX) or valid_digest(value))
     }
     if not pending:
         return params
@@ -180,15 +183,13 @@ def resolve_references(
         for key, text in pending.items():
             if text.startswith(RUN_REF_PREFIX):
                 resolved[key] = _from_run(session, key, text[len(RUN_REF_PREFIX) :])
-            elif valid_digest(text):
+            else:
                 resolved[key] = _from_digest(session, key, text, artifacts)
-            # Anything else is left alone: the type check names it better than
-            # a guess about what was meant would.
     return resolved
 
 
-def _from_run(session: Session, key: str, spelling: str) -> dict[str, Any]:
-    """``<run id>.<output>`` as the reference that run produced."""
+def _from_run(session: Session, key: str, spelling: str) -> Any:
+    """``<run id>.<output>`` as the value that run produced."""
     run_id, _, output = spelling.partition(".")
     if not run_id or not output:
         raise RunRejected(
@@ -199,13 +200,16 @@ def _from_run(session: Session, key: str, spelling: str) -> dict[str, Any]:
     if run is None:
         raise RunRejected(f"Parameter '{key}': there is no run '{run_id}'")
 
-    # The run's own result first: that is the reference as its producer made
-    # it, file name and all. The rows are the fallback, and they carry the
-    # message name instead — which loads the same bytes either way.
-    candidate = (run.result or {}).get(output)
-    if is_reference(candidate):
-        return dict(candidate)
+    # The run's own result first: that is the value as its producer made it,
+    # an artifact's file name and all. Copied, because what comes back is
+    # handed on as this run's parameter and must not alias the other run's row.
+    result = run.result or {}
+    if output in result:
+        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.
     rows = session.exec(
         select(RunArtifact).where(col(RunArtifact.run_id) == run_id)
     ).all()
@@ -217,10 +221,9 @@ def _from_run(session: Session, key: str, spelling: str) -> dict[str, Any]:
                 "media_type": row.media_type or "application/octet-stream",
                 "name": row.name,
             }
-    known = ", ".join(sorted(row.name for row in rows)) or "none"
+    known = ", ".join(sorted({*result, *(row.name for row in rows)})) or "none"
     raise RunRejected(
-        f"Parameter '{key}': run '{run_id}' has no artifact '{output}' "
-        f"(it made: {known})"
+        f"Parameter '{key}': run '{run_id}' has no output '{output}' (it made: {known})"
     )
 
 
@@ -428,9 +431,9 @@ class RunCache:
     def __init__(self, artifacts: ArtifactStore | None = None) -> None:
         self.artifacts = artifacts
 
-    def lookup(self, key: str) -> tuple[bool, dict[str, Any] | None]:
+    def lookup(self, key: str) -> CacheHit | None:
         if not key:
-            return False, None
+            return None
         try:
             with Session(db_engine) as session:
                 rows = session.exec(
@@ -450,11 +453,18 @@ class RunCache:
                     outputs = json.loads(row.outputs or "null")
                     if outputs and self._unrestorable(outputs):
                         continue
-                    return True, outputs
+                    source = session.get(Run, row.run_id)
+                    return CacheHit(
+                        flow=source.flow if source is not None else "",
+                        outputs=outputs,
+                        # A row that was itself restored holds no series, so
+                        # follow its pointer rather than adding a hop to it.
+                        metrics_run=row.cached_from or row.run_id,
+                    )
         except Exception:
             # A cache that cannot answer is a cache miss, never a failed run.
             logger.exception("Cache lookup failed; running the node instead")
-        return False, None
+        return None
 
     def _unrestorable(self, outputs: dict[str, Any]) -> bool:
         return any(
@@ -842,6 +852,7 @@ class RunService:
             duration_ms=outcome.duration_ms,
             error=outcome.error[:ERROR_CAP],
             logs=outcome.logs[:LOG_CAP],
+            cached_from=outcome.cached_from,
             # Together or not at all: a row carrying a key must be one a
             # lookup can actually restore from.
             cache_key=outcome.cache_key if outputs is not None else "",