Key a node on the code it reaches, not on the repository around it

`sync` follows each node function's imports through the project's own modules
— stopping at the standard library, at anything installed, and at Fluksio
itself, whose checkout would otherwise be most of every digest — and records
the file list with what it hashed to. The engine hashes those files again when
the run is claimed, so the fingerprint is live rather than a snapshot, and
falls back to what sync recorded when it cannot see them: a remote worker's
runs used to share one empty digest, and therefore one key.

Three things follow. Editing a helper a node calls into re-runs that node, as
before. Editing something the node never reaches no longer re-runs anything —
a notebook two directories away was invalidating every arm. And
`Run.code_digest` is now the hash of its nodes' digests, so it is neither
looser nor tighter than "the code behind these numbers", which is what makes
it worth joining an exported table on.

`sync` says so too: it compares the per-node digest against the stored one, so
a helper edit prints `train: updated (flow, fit)` instead of `unchanged`. The
digest is read when the document is built rather than when the flow is
declared, so a second `sync()` in one process sees an edit between them.

Also: `fluksio runs` shows only the inputs that differ from what the flow
declares, fitted to the terminal, so a flow taking a few kB of json no longer
wraps every line.

Every existing cache entry misses once — the fingerprint changed shape.

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

backend/fluksio/flow/runs.py

@@ -56,6 +56,7 @@ from fluksio.flow.schemas import FlowDef, NodeDef
 from fluksio.flow.state import MemoryState, StateBackend
 from fluksio.flow.store import FlowStore
 from fluksio.models import Run, RunArtifact, RunMetric, RunNode
+from fluksio.sdk.imports import digest_of as files_digest
 
 logger = logging.getLogger(__name__)
 
@@ -615,12 +616,71 @@ def repo_digest(repo: str) -> str:
     return answer
 
 
+#: What a node's own modules hashed to, and the stat state behind that answer.
+#: Keyed by the node's file list, the same trade `_digests` makes.
+_node_digests: dict[tuple[str, ...], tuple[tuple[tuple[int, int], ...], str]] = {}
+
+
+def node_digest(node: NodeDef, flow: FlowDef) -> str:
+    """What the code this node reaches hashes to now.
+
+    `sync` recorded which project modules the node's function imports its way
+    to, so this is the narrow answer the repository-wide digest could not
+    give: it moves when the node's own code moves and stays put when a
+    notebook two directories away changes.
+
+    Read live, because the shim imports from disk whenever the worker starts
+    and a sweep may sit queued for hours. An engine that cannot see the files
+    keeps what sync recorded instead of nothing — a remote worker's runs used
+    to share one empty digest, and one key.
+    """
+    files = node.code_files
+    if not files:
+        # Drawn on the canvas, or synced before this was recorded. The
+        # repository is the older, blunter answer and still the only one.
+        return code_digest(flow)
+    paths = tuple(sorted(files.values()))
+    state: list[tuple[int, int]] = []
+    for path in paths:
+        try:
+            info = os.stat(path)
+        except OSError:
+            return node.code_digest
+        state.append((info.st_mtime_ns, info.st_size))
+    stamp = tuple(state)
+    cached = _node_digests.get(paths)
+    if cached is not None and cached[0] == stamp:
+        return cached[1]
+    answer = files_digest(files) or node.code_digest
+    _node_digests[paths] = (stamp, answer)
+    return answer
+
+
+def node_digests(flow: FlowDef) -> dict[str, str]:
+    """Every node's digest, which is what its fingerprint is keyed on."""
+    return {node.id: node_digest(node, flow) for node in flow.nodes}
+
+
 def code_digest(flow: FlowDef) -> str:
     """The digest of the repository a code-defined flow was declared in."""
     origin = flow.origin
     return repo_digest(origin.repo) if origin is not None else ""
 
 
+def run_digest(flow: FlowDef) -> str:
+    """What the run records as the code it ran: its nodes' digests, hashed.
+
+    One column over a flow's several nodes, so it changes when any node's own
+    code does and not when anything else in the repository does — which is
+    what makes it worth joining an exported table on.
+    """
+    parts = node_digests(flow)
+    if not any(parts.values()):
+        return ""
+    canonical = json.dumps(parts, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode()).hexdigest()
+
+
 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:
@@ -794,7 +854,7 @@ class RunService:
             flow_version=flow.version,
             commit=self.controller.store.head(),
             origin_commit=origin_commit(flow),
-            code_digest=code_digest(flow),
+            code_digest=run_digest(flow),
             params=params,
             params_digest=digest_of(params, seed),
             seed=seed,
@@ -1070,14 +1130,19 @@ class RunService:
             # run at once, and the code on disk is free to move in the hours
             # before the last of them starts. What the record must name is the
             # state that ran, not the state that was submitted.
-            run.code_digest = self._restamp(run, code_digest(flow))
+            digests = node_digests(flow)
+            run.code_digest = self._restamp(run, run_digest(flow))
             state = self._state_factory(f"{RUN_NAMESPACE}:{run_id}")
             pipeline = self.controller.build_run_pipeline(
                 flow,
                 state=state,
                 observer=observe,
                 emission_observer=sink.handle,
-                run=RunContext(run_id=run_id, code_digest=run.code_digest),
+                run=RunContext(
+                    run_id=run_id,
+                    code_digest=run.code_digest,
+                    node_digests=digests,
+                ),
                 run_cache=None if run.no_cache else self._cache,
             )
             with self._lock: