Stage caching for batch runs, and an engine that lives in the command

A code node in a batch run is now fingerprinted by its source, its raw
settings and the values it reads — an artifact input counting as its digest,
which is what the content addressing was always for. A run that finds the key
restores what the earlier one returned and skips the node, recorded as
`cached`. The run history is the cache: `run_node.outputs` beside the
`cache_key` the schema already had, no second store. On for code nodes, never
for the built-in and connector types that have side effects; off per node with
`@node(cache=False)` and per run with `--no-cache`.

Emissions are not replayed on a hit, so a cached training node returns its
result without redrawing its curve. Recorded in NOTEPAD.md with the two other
deliberate limits.

`fluksio run --local` boots the real app in the command's own process and
drives it through its ASGI interface behind the ordinary client, so a run no
longer needs a `serve` terminal beside it — same data directory, same history,
and the cache carries between the two. It always waits, because the engine it
starts lives exactly as long as the command.

Also: `fluksio sweep --param lr=0.1,0.01` for the product of the lists,
`run --follow` for a run's numbers as they arrive, Ctrl-C cancelling a waited
run rather than abandoning it, coloured statuses on a terminal, and `name`
made optional on the metrics endpoint so a follower can ask for every series.

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

backend/fluksio/flow/runs.py

@@ -43,6 +43,7 @@ from sqlalchemy.dialects.sqlite import insert as upsert
 from sqlmodel import Session, col, select
 
 from fluksio.core.db import engine as db_engine
+from fluksio.flow.artifacts import ArtifactStore, is_reference
 from fluksio.flow.controller import FlowController, RunContext
 from fluksio.flow.messages import qualify
 from fluksio.flow.pipeline import NodeOutcome, Pipeline
@@ -69,6 +70,10 @@ CLAIM_COUNT = 4
 CLAIM_BLOCK_MS = 1000
 ERROR_CAP = 2000
 LOG_CAP = 8000
+#: How much of a node's returned outputs is kept so a later run can restore
+#: them. Past it the node simply is not cacheable — a cache is not a store,
+#: and an artifact is how a large result is meant to travel.
+OUTPUT_CAP = 256_000
 #: Reported numbers held before they are written. A training loop reporting
 #: every step must not be a round trip every step.
 METRIC_BATCH = 500
@@ -288,6 +293,68 @@ def origin_commit(flow: FlowDef) -> str:
     return f"{origin.commit}-dirty" if origin.dirty else origin.commit
 
 
+def _cacheable(outcome: NodeOutcome) -> str | None:
+    """A node's outputs as stored, or None when it may not be reused."""
+    if not outcome.ok or not outcome.cache_key:
+        return None
+    try:
+        text = json.dumps(outcome.output_values, sort_keys=True, separators=(",", ":"))
+    except (TypeError, ValueError):
+        return None
+    return text if len(text) <= OUTPUT_CAP else None
+
+
+class RunCache:
+    """What earlier runs already worked out, looked up by cache key.
+
+    The rows are the run history itself — a node that ran successfully and
+    whose outputs were small enough to keep is a cache entry, without a store
+    of its own. An entry whose artifacts have gone from the content store is
+    not a hit: the reference would name bytes nobody can open.
+    """
+
+    def __init__(self, artifacts: ArtifactStore | None = None) -> None:
+        self.artifacts = artifacts
+
+    def lookup(self, key: str) -> tuple[bool, dict[str, Any] | None]:
+        if not key:
+            return False, None
+        try:
+            with Session(db_engine) as session:
+                rows = session.exec(
+                    select(RunNode)
+                    .where(
+                        col(RunNode.cache_key) == key,
+                        col(RunNode.status).in_(("ok", "cached")),
+                        col(RunNode.outputs).is_not(None),
+                    )
+                    # Run ids are time-ordered, so this is the most recent
+                    # first. A handful is enough to get past entries whose
+                    # artifacts have been collected.
+                    .order_by(col(RunNode.run_id).desc())
+                    .limit(5)
+                ).all()
+                for row in rows:
+                    outputs = json.loads(row.outputs or "null")
+                    if outputs and self._unrestorable(outputs):
+                        continue
+                    return True, outputs
+        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
+
+    def _unrestorable(self, outputs: dict[str, Any]) -> bool:
+        return any(
+            is_reference(value)
+            and (
+                self.artifacts is None
+                or self.artifacts.path(str(value["digest"])) is None
+            )
+            for value in outputs.values()
+        )
+
+
 class RunService:
     """Accepts runs, drives them, and writes down what they did."""
 
@@ -297,9 +364,11 @@ class RunService:
         queue: WorkQueue,
         state_factory: Callable[[str], StateBackend] | None = None,
         parallel: int = MAX_PARALLEL,
+        artifacts: ArtifactStore | None = None,
     ) -> None:
         self.controller = controller
         self.queue = queue
+        self._cache = RunCache(artifacts)
         # Without one, a run gets a private in-memory state — which is exactly
         # the isolation it wants, minus surviving the process.
         self._state_factory = state_factory or (lambda _ns: MemoryState())
@@ -356,6 +425,7 @@ class RunService:
         cause: str = "api",
         actor: str = "",
         draft: bool = False,
+        no_cache: bool = False,
     ) -> Run:
         """Journal a run and wake an engine up for it. Never blocks on it."""
         flow = self.controller.store.read_flow(flow_name, draft=draft)
@@ -378,6 +448,7 @@ class RunService:
             seed=seed,
             group_id=group_id,
             cause=cause,
+            no_cache=no_cache,
             status="queued",
             labels=required_labels(flow),
             created_at=datetime.now(UTC),
@@ -605,6 +676,7 @@ class RunService:
                 observer=observe,
                 emission_observer=sink.handle,
                 run=RunContext(run_id=run_id),
+                run_cache=None if run.no_cache else self._cache,
             )
             with self._lock:
                 self._active[run_id] = pipeline
@@ -644,14 +716,19 @@ class RunService:
                     logger.warning("Could not clear state of run %s", run_id)
 
     def _record_node(self, run_id: str, outcome: NodeOutcome) -> None:
+        outputs = _cacheable(outcome)
         row = RunNode(
             run_id=run_id,
             node=outcome.node[:255],
-            status="ok" if outcome.ok else "error",
+            status="cached" if outcome.cached else ("ok" if outcome.ok else "error"),
             started_at=datetime.now(UTC),
             duration_ms=outcome.duration_ms,
             error=outcome.error[:ERROR_CAP],
             logs=outcome.logs[:LOG_CAP],
+            # 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 "",
+            outputs=outputs,
         )
         try:
             with Session(db_engine) as session: