diff --git a/backend/fluksio/alembic/versions/d1f7a3c8b204_run_node_cached_from.py b/backend/fluksio/alembic/versions/d1f7a3c8b204_run_node_cached_from.py
new file mode 100644
index 0000000..46c1f66
--- /dev/null
+++ b/backend/fluksio/alembic/versions/d1f7a3c8b204_run_node_cached_from.py
@@ -0,0 +1,37 @@
+"""run_node.cached_from
+
+Which run a restored node's values came from. A cache hit replays no emissions,
+so the series stays where it was recorded and the run that reused it points at
+it rather than copying a few thousand rows per reuse.
+
+Revision ID: d1f7a3c8b204
+Revises: c9a5d1e73b48
+Create Date: 2026-08-25
+
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "d1f7a3c8b204"
+down_revision = "c9a5d1e73b48"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "run_node",
+        sa.Column(
+            "cached_from",
+            sqlmodel.sql.sqltypes.AutoString(length=64),
+            nullable=False,
+            server_default="",
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("run_node", "cached_from")
diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py
index c2dc375..ca742b0 100644
--- a/backend/fluksio/api/routes/runs.py
+++ b/backend/fluksio/api/routes/runs.py
@@ -12,9 +12,10 @@ from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel, Field
 from sqlalchemy import func
 from sqlalchemy import select as sa_select
-from sqlmodel import col, select
+from sqlmodel import Session, col, select
 
 from fluksio.api.deps import CurrentUser, SessionDep, get_current_user
+from fluksio.flow.messages import requalify
 from fluksio.flow.runs import RunRejected, RunService, new_run_id
 from fluksio.flow.store import FlowNotFound
 from fluksio.models import Run, RunArtifact, RunMetric, RunNode
@@ -58,6 +59,9 @@ class RunNodeRow(BaseModel):
     #: What this node's result was looked up by. Empty when it may not be
     #: reused; `status` is "cached" when it was.
     cache_key: str = ""
+    #: Which run it was restored from, when it was. That run is also where this
+    #: node's series was recorded.
+    cached_from: str = ""
 
 
 class ArtifactRow(BaseModel):
@@ -284,6 +288,59 @@ async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any:
     return run
 
 
+def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]:
+    """A run's numbers, including the ones a cached node points at.
+
+    A cache hit replays no emissions, so a node restored from an earlier run has
+    no rows of its own — it carries that run's id instead, and its series is read
+    from there. Names are re-qualified on the way out, because the same node
+    reached through two flows publishes under two names and the caller asked for
+    this run's.
+    """
+    statement = select(RunMetric).where(col(RunMetric.run_id) == run_id)
+    if name:
+        statement = statement.where(col(RunMetric.name) == name)
+    rows = list(session.exec(statement))
+
+    restored = session.exec(
+        select(RunNode).where(
+            col(RunNode.run_id) == run_id, col(RunNode.cached_from) != ""
+        )
+    ).all()
+    if restored:
+        run = session.get(Run, run_id)
+        flow = run.flow if run is not None else ""
+        for node_row in restored:
+            source = session.get(Run, node_row.cached_from)
+            if source is None:
+                # The run it came from is gone — deleted with its flow. The
+                # outputs are still on this run; the curve is not recoverable.
+                continue
+            source_node = requalify(node_row.node, flow, source.flow)
+            for row in session.exec(
+                select(RunMetric).where(
+                    col(RunMetric.run_id) == node_row.cached_from,
+                    col(RunMetric.node) == source_node,
+                )
+            ):
+                renamed = requalify(row.name, source.flow, flow)
+                if name and renamed != name:
+                    continue
+                rows.append(
+                    RunMetric(
+                        run_id=run_id,
+                        name=renamed,
+                        step=row.step,
+                        node=node_row.node,
+                        ts=row.ts,
+                        value=row.value,
+                    )
+                )
+
+    rows.sort(key=lambda row: (row.name, row.step))
+    return rows
+
+
 @router.get("/{run_id}/metrics", response_model=list[MetricPoint])
 def read_metrics(
     run_id: str, session: SessionDep, name: str = "", stride: int = 1
@@ -293,11 +350,7 @@ def read_metrics(
     ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart
     is 3000 points nobody can see.
     """
-    statement = select(RunMetric).where(col(RunMetric.run_id) == run_id)
-    if name:
-        statement = statement.where(col(RunMetric.name) == name)
-    statement = statement.order_by(col(RunMetric.name), col(RunMetric.step))
-    rows = list(session.exec(statement))
+    rows = _series(session, run_id, name)
     if stride > 1:
         rows = rows[:: max(1, stride)]
     return rows
@@ -323,11 +376,7 @@ def compare_metric(session: SessionDep, ids: str, metric: str) -> Any:
         run = runs.get(run_id)
         if run is None:
             continue
-        rows = session.exec(
-            select(RunMetric)
-            .where(col(RunMetric.run_id) == run_id, col(RunMetric.name) == metric)
-            .order_by(col(RunMetric.step))
-        ).all()
+        rows = _series(session, run_id, metric)
         label = run_id
         if run.seed is not None:
             label = f"{run_id} (seed {run.seed})"
diff --git a/backend/fluksio/flow/messages.py b/backend/fluksio/flow/messages.py
index 7a91bfb..261e85d 100644
--- a/backend/fluksio/flow/messages.py
+++ b/backend/fluksio/flow/messages.py
@@ -286,3 +286,16 @@ def qualify(flow: str, name: str) -> str:
 def flow_of(qualified: str) -> str:
     """The flow a qualified message name belongs to."""
     return qualified.split(".", 1)[0]
+
+
+def requalify(qualified: str, source_flow: str, target_flow: str) -> str:
+    """A name owned by one flow, read as the same name in another.
+
+    What a node published as ``train.loss`` is ``quick.loss`` when the same node
+    is reached through ``quick``: the value is the same, only the namespace it
+    hangs in differs. A name the source flow does not own is left alone — a node
+    deliberately publishing into another flow's namespace keeps doing so.
+    """
+    if source_flow == target_flow or not qualified.startswith(f"{source_flow}."):
+        return qualified
+    return f"{target_flow}{qualified[len(source_flow) :]}"
diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py
index ab4d9fc..3cb586d 100644
--- a/backend/fluksio/flow/pipeline.py
+++ b/backend/fluksio/flow/pipeline.py
@@ -30,7 +30,7 @@ from pydantic import BaseModel
 from fluksio.flow import logs
 from fluksio.flow.artifacts import is_reference
 from fluksio.flow.events import EventBus
-from fluksio.flow.messages import flow_of
+from fluksio.flow.messages import flow_of, requalify
 from fluksio.flow.nodes import Node
 from fluksio.flow.queue import WorkQueue
 from fluksio.flow.state import MemoryState, StateBackend
@@ -103,6 +103,9 @@ class NodeOutcome(BaseModel):
     artifacts: dict[str, dict[str, Any]] = {}
     #: Restored from an earlier run rather than executed.
     cached: bool = False
+    #: Which run the restored values came from — and, because a restored row
+    #: holds no series of its own, which run this node's metrics live in.
+    cached_from: str = ""
     #: What an equal execution of this node would be looked up by. Empty when
     #: the node is not cacheable at all.
     cache_key: str = ""
@@ -111,6 +114,20 @@ class NodeOutcome(BaseModel):
     output_values: dict[str, Any] | None = None
 
 
+class CacheHit(BaseModel):
+    """An earlier run of a node, as the thing restoring it needs to see."""
+
+    #: The flow whose namespace the stored outputs are named in. A node reached
+    #: through two flows publishes the same values under two names.
+    flow: str = ""
+    #: What it published. None means it published nothing, which is a result
+    #: worth restoring and has to be distinguishable from a miss.
+    outputs: dict[str, Any] | None = None
+    #: The run holding this node's series. Not necessarily the run the outputs
+    #: were read from: a row that was itself restored has no series of its own.
+    metrics_run: str = ""
+
+
 class RunCacheLookup(Protocol):
     """Where a pipeline asks whether a node has already been run.
 
@@ -118,8 +135,8 @@ class RunCacheLookup(Protocol):
     hands it one of these, a test hands it a dict.
     """
 
-    def lookup(self, key: str) -> tuple[bool, dict[str, Any] | None]:
-        """(hit, outputs). Outputs None on a hit means it published nothing."""
+    def lookup(self, key: str) -> CacheHit | None:
+        """What an equal execution produced, or None when there is no entry."""
 
 
 def run_cache_key(fingerprint: str, inputs: dict[str, Any]) -> str:
@@ -817,19 +834,32 @@ class Pipeline:
         The outputs go into state as if the node had just returned them, which
         is what everything downstream reads — a run's state namespace is its
         own, so a skipped node leaves nothing behind for the next one to find.
-        What it emitted on the way is not restored: those values were the
-        story of an execution that is not happening this time.
+        Named for *this* node's flow rather than the one that produced them: the
+        same node reached through two flows publishes the same values under two
+        names, and downstream here looks for the ones this flow owns.
+
+        What it emitted on the way is not restored either — those values were
+        the story of an execution that is not happening. The run it came from is
+        recorded instead, and that is where its series is read from.
         """
         assert self.run_cache is not None
         try:
-            hit, outputs = self.run_cache.lookup(key)
+            hit = self.run_cache.lookup(key)
         except Exception:
             # A cache that cannot answer is a cache miss, never a failed node.
             logger.exception("Cache lookup failed for '%s'", node.id)
             return False, None
-        if not hit:
+        if hit is None:
             return False, None
 
+        outputs = (
+            {
+                requalify(name, hit.flow, node.flow): value
+                for name, value in hit.outputs.items()
+            }
+            if hit.outputs
+            else hit.outputs
+        )
         if outputs:
             self._record_outputs(node, outputs, state)
         self._publish(
@@ -848,6 +878,7 @@ class Pipeline:
                 node=node.id,
                 ok=True,
                 cached=True,
+                cached_from=hit.metrics_run,
                 cache_key=key,
                 outputs=len(outputs or {}),
                 output_values=outputs,
diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py
index 41d015b..a2540d8 100644
--- a/backend/fluksio/flow/runs.py
+++ b/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 "",
diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py
index 557a685..7bbdedc 100644
--- a/backend/fluksio/models.py
+++ b/backend/fluksio/models.py
@@ -397,6 +397,10 @@ class RunNode(SQLModel, table=True):
     #: Everything this node's output depends on, hashed: its source, its
     #: settings and the values it read. What a later run looks itself up by.
     cache_key: str = Field(default="", index=True, max_length=64)
+    #: The run this node's values were restored from, when it was not executed.
+    #: Its series lives there too, which is what makes a cached run's curve
+    #: readable without copying a few thousand rows per reuse.
+    cached_from: str = Field(default="", max_length=64)
     #: What it returned, as canonical JSON, so a node with this key can be
     #: skipped and its outputs restored. None when it may not be reused —
     #: opted out, too large, or a value JSON cannot carry. Written with
diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py
index b6739e4..74ef3c2 100644
--- a/backend/fluksio/sdk/cli.py
+++ b/backend/fluksio/sdk/cli.py
@@ -282,6 +282,12 @@ def cmd_sync(args: argparse.Namespace) -> int:
 
 
 def _coerce(value: str, dtype: str) -> Any:
+    if value.startswith("@run:") or value.startswith("sha256:"):
+        # A run's output, named rather than typed out. Sent as it stands: the
+        # engine turns it into the value itself, whatever type that is. Passing
+        # the whole thing as JSON still works, and is what a script that
+        # already has it in hand would do.
+        return value
     if dtype == "int":
         return int(value)
     if dtype == "float":
@@ -290,13 +296,6 @@ def _coerce(value: str, dtype: str) -> Any:
         return value.lower() in ("true", "1", "yes", "on")
     if dtype == "str":
         return value
-    if dtype == "artifact" and (
-        value.startswith("@run:") or value.startswith("sha256:")
-    ):
-        # The engine turns these into the reference itself. Passing the whole
-        # object as JSON still works, and is what a script that already has one
-        # would do.
-        return value
     return json.loads(value)
 
 
diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py
index 1a7a1ca..5059b5c 100644
--- a/backend/tests/api/routes/test_runs.py
+++ b/backend/tests/api/routes/test_runs.py
@@ -8,7 +8,7 @@ import json
 from datetime import UTC, datetime
 
 import pytest
-from sqlmodel import Session
+from sqlmodel import Session, select
 
 from fluksio.core.config import settings
 from fluksio.core.db import engine as db_engine
@@ -25,7 +25,7 @@ from fluksio.flow.runs import (
     seed_values,
 )
 from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef
-from fluksio.models import Run, RunArtifact, RunNode
+from fluksio.models import Run, RunArtifact, RunMetric, RunNode
 
 
 def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path):
@@ -66,13 +66,17 @@ def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path):
         session.commit()
 
     cache = RunCache(store)
-    assert cache.lookup(plain) == (True, {"study.loss": 1.5})
-    assert cache.lookup(with_artifact) == (True, {"study.data": reference})
+    hit = cache.lookup(plain)
+    assert hit is not None and hit.outputs == {"study.loss": 1.5}
+    # Where its series is, so a run reusing it can read the curve back.
+    assert hit.metrics_run == "cache-1"
+    found = cache.lookup(with_artifact)
+    assert found is not None and found.outputs == {"study.data": reference}
     # Its bytes have gone from the store, so the reference names nothing a
     # restored run could open. That is a miss, not a broken run.
-    assert cache.lookup(collected) == (False, None)
-    assert cache.lookup("never-seen") == (False, None)
-    assert cache.lookup("") == (False, None)
+    assert cache.lookup(collected) is None
+    assert cache.lookup("never-seen") is None
+    assert cache.lookup("") is None
 
 
 def test_what_may_be_stored_as_a_cache_entry():
@@ -199,6 +203,65 @@ def test_a_reference_passed_whole_is_left_alone(made_artifact):
     }
 
 
+def chaining_flow() -> FlowDef:
+    """A flow taking a json config and a label another run worked out."""
+    meta = MessageSpec(name="meta", dtype=DType.JSON)
+    label = MessageSpec(name="label", dtype=DType.STR)
+    return FlowDef(
+        name="study",
+        mode="batch",
+        inputs=[FlowInput(spec=meta), FlowInput(spec=label)],
+        nodes=[NodeDef(id="train", requires=[meta, label])],
+    )
+
+
+@pytest.fixture
+def made_config():
+    """A finished run whose result is an object, not bytes."""
+    meta = {"rows": 256, "source": "builtin"}
+    run_id = new_run_id()
+    with Session(db_engine) as session:
+        session.add(
+            Run(
+                id=run_id,
+                flow="generate",
+                status="ok",
+                result={"meta": meta, "label": "run-7"},
+                created_at=datetime.now(UTC),
+            )
+        )
+        session.commit()
+    yield run_id, meta
+    with Session(db_engine) as session:
+        session.delete(session.get(Run, run_id))
+        session.commit()
+
+
+def test_a_json_input_may_name_a_run_s_output(made_config):
+    """The gap this closes: chaining without pasting the object into a shell."""
+    run_id, meta = made_config
+    flow = chaining_flow()
+    resolved = resolve_references(flow, {"meta": f"@run:{run_id}.meta"})
+
+    assert resolved["meta"] == meta
+    # And it is the value's own type from here on, so the input check passes.
+    assert seed_values(flow, resolved)["study.meta"] == meta
+
+
+def test_the_spelling_is_reserved_on_a_text_input_too(made_config):
+    run_id, _meta = made_config
+    resolved = resolve_references(chaining_flow(), {"label": f"@run:{run_id}.label"})
+
+    assert resolved["label"] == "run-7"
+
+
+def test_text_that_names_nothing_is_still_left_alone(made_config):
+    """Only the two spellings are read as names; everything else is a value."""
+    _run_id, _meta = made_config
+    params = {"label": "@run-of-the-mill", "meta": {"rows": 1}}
+    assert resolve_references(chaining_flow(), params) == params
+
+
 def test_the_overview_counts_a_flow_the_list_page_would_not_reach(
     client, superuser_token_headers
 ):
@@ -231,3 +294,89 @@ def test_overview_is_not_read_as_a_run_id(client, superuser_token_headers):
 
     assert answer.status_code == 200
     assert isinstance(answer.json(), list)
+
+
+# -----------------------------------------------------------------------------
+# A cached node's curve
+#
+# A hit replays no emissions, so the series stays in the run that recorded it
+# and the run reusing it points there. Reading either one answers the same.
+# -----------------------------------------------------------------------------
+
+
+@pytest.fixture
+def reused_run():
+    """A run of `quick` whose node was restored from a run of `train`."""
+    with Session(db_engine) as session:
+        made = datetime.now(UTC)
+        session.add(Run(id="src-1", flow="train", status="ok", created_at=made))
+        session.add(Run(id="reuse-1", flow="quick", status="ok", created_at=made))
+        session.add(RunNode(run_id="src-1", node="train.fit", status="ok"))
+        session.add(
+            RunNode(
+                run_id="reuse-1",
+                node="quick.fit",
+                status="cached",
+                cached_from="src-1",
+            )
+        )
+        for step, value in enumerate([3.0, 2.0, 1.0]):
+            session.add(
+                RunMetric(
+                    run_id="src-1",
+                    name="train.loss",
+                    step=step,
+                    node="train.fit",
+                    value=value,
+                )
+            )
+        session.commit()
+    yield
+    with Session(db_engine) as session:
+        for row in session.exec(select(RunMetric)).all():
+            session.delete(row)
+        for row in session.exec(select(RunNode)).all():
+            session.delete(row)
+        for run_id in ("src-1", "reuse-1"):
+            run = session.get(Run, run_id)
+            if run is not None:
+                session.delete(run)
+        session.commit()
+
+
+@pytest.mark.usefixtures("reused_run")
+def test_a_cached_node_answers_with_the_curve_it_was_restored_from(
+    client, superuser_token_headers
+):
+    points = client.get(
+        f"{settings.API_V1_STR}/runs/reuse-1/metrics",
+        headers=superuser_token_headers,
+    ).json()
+
+    # Named for the flow that asked, not the one that recorded it.
+    assert [point["name"] for point in points] == ["quick.loss"] * 3
+    assert [point["value"] for point in points] == [3.0, 2.0, 1.0]
+
+    named = client.get(
+        f"{settings.API_V1_STR}/runs/reuse-1/metrics",
+        params={"name": "quick.loss"},
+        headers=superuser_token_headers,
+    ).json()
+    assert len(named) == 3
+
+
+@pytest.mark.usefixtures("reused_run")
+def test_a_curve_whose_run_is_gone_is_empty_rather_than_an_error(
+    client, superuser_token_headers
+):
+    """Deleting a flow deletes its runs; what pointed at one is left holding it."""
+    with Session(db_engine) as session:
+        session.delete(session.get(Run, "src-1"))
+        session.commit()
+
+    answer = client.get(
+        f"{settings.API_V1_STR}/runs/reuse-1/metrics",
+        headers=superuser_token_headers,
+    )
+    assert answer.status_code == 200
+    assert answer.json() == []
diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py
index 7ed5fc9..973c196 100644
--- a/backend/tests/flow/test_runs.py
+++ b/backend/tests/flow/test_runs.py
@@ -11,7 +11,7 @@ import pytest
 from fluksio.flow.events import EventBus
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.nodes import Node
-from fluksio.flow.pipeline import NodeOutcome, Pipeline, run_cache_key
+from fluksio.flow.pipeline import CacheHit, NodeOutcome, Pipeline, run_cache_key
 from fluksio.flow.runs import (
     MetricSink,
     RunRejected,
@@ -351,15 +351,20 @@ def test_a_flow_without_a_seed_input_ignores_the_runs_seed():
 class FakeCache:
     """A stage cache with no database behind it, and a record of what it was asked."""
 
-    def __init__(self, entries: dict[str, dict | None] | None = None) -> None:
+    def __init__(
+        self, entries: dict[str, dict | None] | None = None, flow: str = "study"
+    ) -> None:
         self.entries = entries or {}
+        self.flow = flow
         self.asked: list[str] = []
 
     def lookup(self, key: str):
         self.asked.append(key)
         if key in self.entries:
-            return True, self.entries[key]
-        return False, None
+            return CacheHit(
+                flow=self.flow, outputs=self.entries[key], metrics_run="earlier"
+            )
+        return None
 
 
 def counting_node(flow: str = "study") -> tuple[Node, list[int]]:
@@ -394,6 +399,30 @@ def test_a_cache_hit_restores_the_outputs_without_running_the_node():
     # downstream of it looks — its namespace holds nothing otherwise.
     assert collect_result(flow, state) == {"loss": 99.0}
     assert seen[0].cached and seen[0].cache_key == key
+    # Where its series is, since the hit replayed none of it.
+    assert seen[0].cached_from == "earlier"
+
+
+def test_a_hit_from_another_flow_restores_under_this_flow_s_names():
+    """The same node reached through two flows publishes under two names."""
+    flow = double_flow()
+    node, calls = counting_node()
+    state = MemoryState()
+    seen: list[NodeOutcome] = []
+    key = run_cache_key("fp-train", {"study.lr": 0.5})
+    # Recorded by a run of "other", which is what a node shared between two
+    # flows gets a hit from — the values are the same, the namespace is not.
+    cache = FakeCache({key: {"other.loss": 99.0}}, flow="other")
+
+    Pipeline(nodes=[node], state=state, observer=seen.append, run_cache=cache).run(
+        seed_values(flow, {"lr": 0.5})
+    )
+
+    assert calls == []
+    assert collect_result(flow, state) == {"loss": 99.0}
+    # And stored that way too, so the next run to reuse this one finds names
+    # it can requalify from a flow that really did publish them.
+    assert seen[0].output_values == {"study.loss": 99.0}
 
 
 def test_a_miss_runs_the_node_and_carries_what_would_be_stored():
diff --git a/docs/code/cli.md b/docs/code/cli.md
index b0c6f59..260ee61 100644
--- a/docs/code/cli.md
+++ b/docs/code/cli.md
@@ -193,19 +193,21 @@ is labelled with is the number your code actually drew from, instead of merely
 looking like it. A flow that declares no such input still records it, and
 nothing reads it. Sweep over seeds with `--param seed=1,2,3`.
 
-An input declared as an `artifact` takes the file a previous run produced,
-named rather than typed out:
+Any input takes what a previous run produced, named rather than typed out —
+a checkpoint, but equally a config object nobody wants to paste into a shell:
 
 ```sh
 fluksio run evaluate --dataset @run:1758042000123-9f2ab41c.dataset
 fluksio run evaluate --dataset sha256:6dd1f0…
+fluksio run train --meta @run:1758042000123-9f2ab41c.dataset_meta
 ```
 
-`@run:<id>.<output>` is what that run's output was, and a bare digest is the
-content itself; the engine resolves either into the reference. Passing the
-whole reference as JSON still works and is what a script that already holds one
-does — which is the same thing `flow.submit(dataset=run.result["dataset"])`
-does from Python.
+`@run:<id>.<output>` is whatever that run's output was, whole and with its own
+type; a bare digest is the content itself, resolved into a reference. Both
+spellings are reserved on every input, `str` included, so an input that has to
+carry one of them literally cannot. Passing the value as JSON still works and
+is what a script that already holds one does — the same thing
+`flow.submit(dataset=run.result["dataset"])` does from Python.
 
 Run a flow with no parameters at a terminal and it asks for them, one line per
 declared input, with the declared value in brackets:
diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md
index 57ab0b7..32c3f59 100644
--- a/docs/concepts/runs.md
+++ b/docs/concepts/runs.md
@@ -202,11 +202,13 @@ canvas and the API can change it too — or for one run with
 `fluksio run --no-cache`, `fluksio sweep --no-cache`, or `"no_cache": true` in
 the submission body.
 
-What a cached node does not bring back is what it emitted on the way. Its
-returned outputs are restored; the values it published mid-execution are not,
-because those were the story of an execution that is not happening this time.
-So a skipped training node contributes no loss curve to the new run — if you
-want the curve, that run has to actually train.
+A cached node replays no emissions — those values were the story of an
+execution that is not happening this time — so its series is not rewritten
+either. The run it was restored from is recorded instead, and that is where the
+curve is read back from: asking the reusing run for its metrics answers with
+the same points, under its own flow's names. The one way to be left with a
+result and no curve is for that earlier run to have been deleted, which
+deleting its flow does.
 
 ## Objects that cannot be serialized
 
@@ -331,10 +333,10 @@ sweep, or specific runs. It re-reads on its own and whenever a run finishes.
 
 ### When a run draws nothing
 
-A node restored from the [stage cache](#stage-caching) replays no emissions —
-a cache hit returns what the node returned, not what it emitted on the way. So
-a run that reused an earlier one has a result and an empty curve, and the
-chart says so rather than looking broken.
+A node restored from the [stage cache](#stage-caching) has its curve read back
+from the run that recorded it. Delete that run — deleting its flow does — and
+the reusing run is left with a result and an empty curve, and the chart says so
+rather than looking broken.
 
 ## What this costs, compared
 
diff --git a/docs/reference/payload-types.md b/docs/reference/payload-types.md
index d8fbe58..4e69969 100644
--- a/docs/reference/payload-types.md
+++ b/docs/reference/payload-types.md
@@ -99,12 +99,6 @@ stays valid wherever the store is reachable from, including on another machine.
 Node code produces one with `fluksio.save_artifact` and opens one with
 `fluksio.load_artifact`. See [Writing node code](../code/nodes.md#bytes-artifacts).
 
-As a *run parameter* it is also accepted as text, since nobody wants to paste
-the object into a shell: `@run:<id>.<output>` names what a run produced, and a
-bare `sha256:…` digest names the content. Both resolve to the reference above
-before the run starts, so the CLI, the run dialog and a python caller all mean
-the same thing.
-
 ### `json`
 
 Anything JSON-serializable. The escape hatch, and the right answer when a
@@ -113,6 +107,18 @@ payload genuinely has no fixed shape.
 Reach for it last. A `json` port tells the canvas, the widget picker and the
 next author nothing.
 
+## Naming a run's output
+
+Any *run parameter* is also accepted as text, since nobody wants to paste an
+object into a shell. `@run:<id>.<output>` names what an earlier run produced —
+whatever its type, an `artifact` reference or a `json` config alike — and a bare
+`sha256:…` digest names content in the artifact store. Both resolve before the
+run starts, so the CLI, the run dialog and a python caller all mean the same
+thing by the same string.
+
+Both spellings are 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.
+
 ## What a widget will bind to
 
 | Widget | Accepts |
diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts
index 2c4bf04..1f452f7 100644
--- a/frontend/src/client/schemas.gen.ts
+++ b/frontend/src/client/schemas.gen.ts
@@ -2541,6 +2541,11 @@ export const RunNodeRowSchema = {
             type: 'string',
             title: 'Cache Key',
             default: ''
+        },
+        cached_from: {
+            type: 'string',
+            title: 'Cached From',
+            default: ''
         }
     },
     type: 'object',
diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts
index b4a95fa..72b7cf6 100644
--- a/frontend/src/client/types.gen.ts
+++ b/frontend/src/client/types.gen.ts
@@ -904,6 +904,7 @@ export type RunNodeRow = {
     error: string;
     logs: string;
     cache_key?: string;
+    cached_from?: string;
 };
 
 export type RunPage = {
diff --git a/frontend/src/components/Flow/FlowBoundary.tsx b/frontend/src/components/Flow/FlowBoundary.tsx
index afeae03..cb2116a 100644
--- a/frontend/src/components/Flow/FlowBoundary.tsx
+++ b/frontend/src/components/Flow/FlowBoundary.tsx
@@ -34,20 +34,16 @@ import { ValuePreview } from "./ValuePreview"
  */
 export function parseByDtype(dtype: DType | undefined, raw: string): unknown {
   if (raw === "") return null
+  // A run's output is named rather than typed out: "@run:<id>.<output>", or a
+  // digest naming bytes. The engine resolves either into the value itself,
+  // whatever its type, so both spellings are kept as text on the way out.
+  if (raw.startsWith("@run:") || raw.startsWith("sha256:")) return raw
   if (dtype === "int" || dtype === "float") {
     const parsed = Number(raw)
     return Number.isNaN(parsed) ? raw : parsed
   }
   if (dtype === "bool") return raw === "true"
   if (dtype === "str") return raw
-  // An artifact is named rather than typed out: "@run:<id>.<output>" or the
-  // digest itself, which the engine resolves into the reference.
-  if (
-    dtype === "artifact" &&
-    (raw.startsWith("@run:") || raw.startsWith("sha256:"))
-  ) {
-    return raw
-  }
   try {
     return JSON.parse(raw)
   } catch {
diff --git a/frontend/src/components/Flow/RunDialog.tsx b/frontend/src/components/Flow/RunDialog.tsx
index 448bd08..329b015 100644
--- a/frontend/src/components/Flow/RunDialog.tsx
+++ b/frontend/src/components/Flow/RunDialog.tsx
@@ -101,7 +101,7 @@ export function RunDialog({
                     placeholder={
                       dtype === "artifact"
                         ? "@run:<id>.<output> or sha256:…"
-                        : (dtype ?? "float")
+                        : `${dtype ?? "float"} or @run:<id>.<output>`
                     }
                     className="text-sm"
                     onChange={(event) =>
diff --git a/frontend/src/components/Runs/RunDetail.tsx b/frontend/src/components/Runs/RunDetail.tsx
index 30c32e6..65beee8 100644
--- a/frontend/src/components/Runs/RunDetail.tsx
+++ b/frontend/src/components/Runs/RunDetail.tsx
@@ -46,8 +46,9 @@ export function RunDetail({ id }: { id: string }) {
   const nodes = run.nodes ?? []
   const artifacts = run.artifacts ?? []
   const reason = statusReason(run)
-  // A run whose nodes were all restored emits nothing, so an empty chart is
-  // the expected outcome rather than a fault. Said once, where it applies.
+  // A restored node's curve comes from the run it was restored from, so an
+  // empty chart here means that run is gone rather than that this one failed.
+  // Said once, where it applies.
   const cached = nodes.some((node) => node.status === "cached")
 
   return (
diff --git a/frontend/src/components/Runs/queries.ts b/frontend/src/components/Runs/queries.ts
index 4f332e2..b278fab 100644
--- a/frontend/src/components/Runs/queries.ts
+++ b/frontend/src/components/Runs/queries.ts
@@ -136,12 +136,13 @@ export async function downloadArtifact(digest: string, name: string) {
 /**
  * Why a finished run can have nothing to draw.
  *
- * A cache hit restores what a node returned, not the values it emitted along
- * the way, so a run whose training node was reused has a result and no curve.
- * Said out loud rather than left as an empty chart, which reads as a fault.
+ * A cache hit replays no emissions, so a restored node's curve is read back
+ * from the run that recorded it. Deleting that run — deleting its flow does —
+ * takes the curve with it, and then a reused run has a result and nothing to
+ * draw. Said out loud rather than left as an empty chart, which reads as a fault.
  */
 export const NO_CURVE =
-  "No curve was recorded. A node restored from the cache replays no emissions, so a run that reused an earlier one draws nothing here — its outputs are still on the result."
+  "No curve was recorded. A node restored from the cache is read back from the run that produced it, so this draws nothing once that run has been deleted — its outputs are still on the result."
 
 /** A run id, short enough for a table cell. The tail is the random half. */
 export const shortId = (id: string) => id.slice(-8)