diff --git a/backend/fluksio/alembic/versions/a4d9e2b71c68_run_code_digest.py b/backend/fluksio/alembic/versions/a4d9e2b71c68_run_code_digest.py
new file mode 100644
index 0000000..fad451f
--- /dev/null
+++ b/backend/fluksio/alembic/versions/a4d9e2b71c68_run_code_digest.py
@@ -0,0 +1,38 @@
+"""run.code_digest
+
+`origin_commit` names a commit, and a dirty tree has the same one all day. The
+node bodies are import shims, so what actually ran is whatever was on disk when
+the worker started. This column hashes those files at run time, which is the
+only thing that tells two runs of an uncommitted tree apart.
+
+Revision ID: a4d9e2b71c68
+Revises: f2c6a8d15e93
+Create Date: 2026-08-26
+
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "a4d9e2b71c68"
+down_revision = "f2c6a8d15e93"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "run",
+        sa.Column(
+            "code_digest",
+            sqlmodel.sql.sqltypes.AutoString(length=64),
+            nullable=False,
+            server_default="",
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("run", "code_digest")
diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py
index ec72429..3d3be6e 100644
--- a/backend/fluksio/api/routes/flows.py
+++ b/backend/fluksio/api/routes/flows.py
@@ -529,7 +529,10 @@ def read_node_source(
             return NodeSource(code=controller.store.read_lib_source(ref))
         except LibNotFound:
             raise HTTPException(status_code=404, detail=f"No shared node '{ref}'")
-    return NodeSource(code=controller.store.read_node_source(name, node_id, draft=True))
+    return NodeSource(
+        code=controller.store.read_node_source(name, node_id, draft=True),
+        missing=not controller.store.has_node_source(name, node_id, draft=True),
+    )
 
 
 @router.put("/{name}/nodes/{node_id}/source", response_model=NodeStatusPublic)
diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py
index 3eb5c58..dc46000 100644
--- a/backend/fluksio/api/routes/runs.py
+++ b/backend/fluksio/api/routes/runs.py
@@ -107,6 +107,10 @@ class RunRow(BaseModel):
     #: decorators. Empty for one drawn on the canvas, where `commit` is the
     #: whole answer to what produced the number.
     origin_commit: str = ""
+    #: What that repository's python files hashed to when the run started.
+    #: Two runs of one dirty tree share a commit and differ here, which is the
+    #: only way to tell apart what they actually executed.
+    code_digest: str = ""
     #: The flow store's own commit. Short, unlike `result`, so the list
     #: carries it: "what code produced this" is a question asked of a table.
     commit: str = ""
diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py
index 4883eb9..f321936 100644
--- a/backend/fluksio/flow/controller.py
+++ b/backend/fluksio/flow/controller.py
@@ -150,6 +150,10 @@ class RunContext:
     """
 
     run_id: str
+    #: What the code-defined flow's repository hashed to when this run started.
+    #: Part of every node's fingerprint, so editing a function the node calls
+    #: into invalidates the cache the way editing the node itself does.
+    code_digest: str = ""
 
 
 @dataclass(frozen=True, slots=True)
@@ -919,9 +923,30 @@ class FlowController:
                     # The raw params, not the resolved ones: a secret's value
                     # must not end up in a key, and its name is what changes
                     # when the node is reconfigured anyway.
+                    #
+                    # `source` is the stored body, which for a code-defined
+                    # flow is a shim that imports the real function — it does
+                    # not move when that function does, and it says nothing
+                    # about what the function calls into. `code` is what
+                    # covers both: the digest of the repository the shim
+                    # imports from. The ports are here because declaring a new
+                    # one changes what the node answers with, which a hit
+                    # would otherwise restore without.
                     node.fingerprint = hashlib.sha256(
                         json.dumps(
-                            {"source": code, "params": node_def.params},
+                            {
+                                "source": code,
+                                "params": node_def.params,
+                                "requires": [
+                                    spec.model_dump(mode="json")
+                                    for spec in node_def.requires
+                                ],
+                                "provides": [
+                                    spec.model_dump(mode="json")
+                                    for spec in node_def.provides
+                                ],
+                                "code": run.code_digest,
+                            },
                             sort_keys=True,
                             separators=(",", ":"),
                         ).encode()
diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py
index 2ab5827..a984bb5 100644
--- a/backend/fluksio/flow/runs.py
+++ b/backend/fluksio/flow/runs.py
@@ -38,6 +38,7 @@ import uuid
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime, timedelta
+from pathlib import Path
 from typing import Any
 
 from sqlalchemy import update
@@ -409,6 +410,77 @@ def origin_commit(flow: FlowDef) -> str:
     return f"{origin.commit}-dirty" if origin.dirty else origin.commit
 
 
+#: What a repository's python files hashed to, and the stat state that answer
+#: was read from. Keyed by path: the walk is cheap, opening every file is not.
+_digests: dict[str, tuple[tuple[tuple[str, int, int], ...], str]] = {}
+
+#: Directories a project's own code is never in, and which are large.
+_SKIP_DIRS = frozenset({"__pycache__", "node_modules"})
+
+
+def repo_digest(repo: str) -> str:
+    """What the python files under ``repo`` currently hash to.
+
+    A code-defined node is stored as a shim that imports the real function, so
+    the engine runs whatever is on disk when the worker starts. The commit
+    cannot name that: an uncommitted tree stamps every run `-dirty`, however
+    many times the code changed underneath. This can — it is read at run time,
+    from the files themselves.
+
+    Empty when there is no repository to read here, which is the honest answer
+    on an engine that does not share a filesystem with the author: a stamp
+    without a digest says the engine could not see the code, and the cache key
+    falls back to what it keyed on before.
+    """
+    if not repo or not os.path.isdir(repo):
+        return ""
+    state: list[tuple[str, int, int]] = []
+    for root, dirs, files in os.walk(repo):
+        dirs[:] = [
+            d
+            for d in dirs
+            if not d.startswith(".")
+            and d not in _SKIP_DIRS
+            # A virtualenv carries this, and its site-packages is not the
+            # project's code — nor is it small.
+            and not os.path.exists(os.path.join(root, d, "pyvenv.cfg"))
+        ]
+        for name in files:
+            if not name.endswith(".py"):
+                continue
+            path = os.path.join(root, name)
+            try:
+                info = os.stat(path)
+            except OSError:
+                continue
+            state.append((os.path.relpath(path, repo), info.st_mtime_ns, info.st_size))
+    state.sort()
+    stamp = tuple(state)
+    cached = _digests.get(repo)
+    if cached is not None and cached[0] == stamp:
+        return cached[1]
+
+    digest = hashlib.sha256()
+    for relative, _mtime, _size in state:
+        try:
+            body = Path(repo, relative).read_bytes()
+        except OSError:
+            continue
+        digest.update(relative.encode())
+        digest.update(b"\0")
+        digest.update(body)
+        digest.update(b"\0")
+    answer = digest.hexdigest()
+    _digests[repo] = (stamp, answer)
+    return answer
+
+
+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 _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:
@@ -579,6 +651,7 @@ class RunService:
             flow_version=flow.version,
             commit=self.controller.store.head(),
             origin_commit=origin_commit(flow),
+            code_digest=code_digest(flow),
             params=params,
             params_digest=digest_of(params, seed),
             seed=seed,
@@ -803,6 +876,17 @@ class RunService:
                 return None
             return session.exec(select(Run).where(col(Run.id) == run_id)).first()
 
+    def _restamp(self, run: Run, digest: str) -> str:
+        """Record the code this run is about to execute, if it moved."""
+        if digest == run.code_digest:
+            return digest
+        with Session(db_engine) as session:
+            session.exec(
+                update(Run).where(col(Run.id) == run.id).values(code_digest=digest)
+            )
+            session.commit()
+        return digest
+
     def _drive(self, run_id: str) -> None:
         run = self._claim(run_id)
         if run is None:
@@ -823,13 +907,18 @@ class RunService:
         sink = MetricSink(run_id)
         try:
             flow = self.controller.store.read_flow(run.flow)
+            # Read again, now that it is this run's turn: a sweep queues every
+            # 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))
             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),
+                run=RunContext(run_id=run_id, code_digest=run.code_digest),
                 run_cache=None if run.no_cache else self._cache,
             )
             with self._lock:
diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py
index 78b3395..eace210 100644
--- a/backend/fluksio/flow/schemas.py
+++ b/backend/fluksio/flow/schemas.py
@@ -158,6 +158,10 @@ class NodeSource(BaseModel):
     """The Python source of a node."""
 
     code: str
+    #: True when nothing is stored and `code` is the new-node template. An
+    #: editor opens on it either way; a client deciding whether somebody wrote
+    #: that code needs to know it was nobody. Read-only — set on the way out.
+    missing: bool = False
 
 
 Health = Literal["ok", "degraded", "down"]
diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py
index 9e33da7..4f857d7 100644
--- a/backend/fluksio/flow/store.py
+++ b/backend/fluksio/flow/store.py
@@ -549,6 +549,18 @@ class FlowStore:
         path = self._node_file(flow, node_id)
         return path.read_text() if path.exists() else DEFAULT_SOURCE
 
+    def has_node_source(self, flow: str, node_id: str, draft: bool = False) -> bool:
+        """Whether a body was ever stored for this node.
+
+        `read_node_source` answers a template when there is none, which is what
+        a new node's editor should open with — but it is a placeholder, not
+        code anybody wrote, and telling the two apart is the difference between
+        "someone edited this" and "this was never written".
+        """
+        if draft and self._draft_node_file(flow, node_id).exists():
+            return True
+        return self._node_file(flow, node_id).exists()
+
     def write_node_source(
         self, flow: str, node_id: str, code: str, draft: bool = False
     ) -> bool:
diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py
index 6116bd4..3f8e8d7 100644
--- a/backend/fluksio/models.py
+++ b/backend/fluksio/models.py
@@ -331,6 +331,12 @@ class Run(SQLModel, table=True):
     #: one names the code it imported. A `-dirty` suffix is git's own way of
     #: saying the tree had changes that are in no commit at all.
     origin_commit: str = Field(default="", max_length=80)
+    #: sha256 over that repository's python files, read when the run actually
+    #: started. The commit cannot tell two runs of an uncommitted tree apart —
+    #: both stamp `-dirty` — and the shim imports whatever is on disk, so this
+    #: is the half that names the code. Empty when the engine cannot see the
+    #: repository, or for a flow drawn on the canvas.
+    code_digest: str = Field(default="", max_length=64)
     params: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict)
     #: sha256 over params and seed. Two runs of the same thing share it, which
     #: is what makes "have I already run this?" a lookup.
diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py
index 4254477..4a00537 100644
--- a/backend/fluksio/sdk/cli.py
+++ b/backend/fluksio/sdk/cli.py
@@ -695,6 +695,19 @@ def cmd_status(args: argparse.Namespace) -> int:
 PARAMS_WIDTH = 80
 
 
+def _stamp(row: dict[str, Any]) -> str:
+    """What code a run ran: the commit, whether it was dirty, and the digest.
+
+    The digest is the part that separates two runs of one uncommitted tree —
+    the commit says `-dirty` for both, however much changed in between.
+    """
+    commit, dirty, _ = (row.get("origin_commit") or "").partition("-dirty")
+    digest = row.get("code_digest") or ""
+    return (
+        commit[:7] + ("-dirty" if dirty else "") + (f"+{digest[:7]}" if digest else "")
+    )
+
+
 def cmd_runs(args: argparse.Namespace) -> int:
     try:
         with _client_for(args) as client:
@@ -704,13 +717,12 @@ def cmd_runs(args: argparse.Namespace) -> int:
     except httpx.HTTPError as exc:
         return _unreachable(exc)
     for row in rows:
-        commit = (row.get("origin_commit") or "")[:7]
         params = json.dumps(row["params"])
         if len(params) > PARAMS_WIDTH:
             params = params[: PARAMS_WIDTH - 3] + "..."
         _say(
             f"{row['id']}  {_status(row['status'], 9)} {row['flow']:<16} "
-            f"{row['duration_ms'] / 1000:7.1f}s  {commit:<8} {params}"
+            f"{row['duration_ms'] / 1000:7.1f}s  {_stamp(row):<22} {params}"
         )
     return 0
 
diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py
index 357c676..138097c 100644
--- a/backend/fluksio/sdk/client.py
+++ b/backend/fluksio/sdk/client.py
@@ -237,8 +237,17 @@ class Client:
         return result
 
     def get_source(self, flow: str, node: str) -> str:
-        result = self._call("GET", f"/flows/{flow}/nodes/{node}/source")
-        return str(result.get("code", ""))
+        return str(self.get_source_entry(flow, node).get("code", ""))
+
+    def get_source_entry(self, flow: str, node: str) -> dict[str, Any]:
+        """A node's stored body, and whether there is one at all.
+
+        `missing` is what separates a node nobody has written from a node
+        somebody edited: the engine answers a new-node template for both, and
+        only one of them is work worth refusing to overwrite.
+        """
+        result: dict[str, Any] = self._call("GET", f"/flows/{flow}/nodes/{node}/source")
+        return result or {}
 
     def put_source(self, flow: str, node: str, code: str) -> dict[str, Any]:
         result: dict[str, Any] = self._call(
@@ -623,13 +632,16 @@ def _sync_one(
     report = SyncReport(target.name)
     stored = client.get_flow(target.name)
     version = 1
+    # What the drift check already read, so the write loop below does not ask
+    # the engine for every node's body a second time.
+    known: dict[str, str] = {}
     if stored is None:
         report.created = True
     else:
         definition = stored.get("definition") or {}
         version = int(definition.get("version") or 1)
         if not force:
-            _refuse_on_drift(client, target, definition)
+            known = _refuse_on_drift(client, target, definition)
 
     saved = client.put_flow(target.document(origin) | {"version": version})
     stored_version = int((saved.get("definition") or {}).get("version") or version)
@@ -640,8 +652,12 @@ def _sync_one(
     version = stored_version
 
     for node_id, code in target.shims().items():
-        if not report.created and client.get_source(target.name, node_id) == code:
-            continue
+        if not report.created:
+            stored_code = known.get(node_id)
+            if stored_code is None:
+                stored_code = client.get_source(target.name, node_id)
+            if stored_code == code:
+                continue
         client.put_source(target.name, node_id, code)
         report.changed.append(node_id)
 
@@ -651,26 +667,43 @@ def _sync_one(
     return report
 
 
-def _refuse_on_drift(client: Client, target: Flow, definition: dict[str, Any]) -> None:
+def _refuse_on_drift(
+    client: Client, target: Flow, definition: dict[str, Any]
+) -> dict[str, str]:
     """Stop before overwriting work that was done somewhere else.
 
     Two ways a stored flow is not ours to replace: it was drawn on the canvas
     and has no origin at all, or one of its node bodies no longer carries the
     line saying it was generated — which means somebody edited the code there.
+
+    A node with *no* stored body is neither. That is what a sync interrupted
+    part of the way through leaves, and it used to read as a canvas edit,
+    because the engine answers the new-node template for it and the template
+    carries no marker either. Re-running the sync is the fix for that state,
+    so refusing to is precisely wrong.
+
+    Returns the bodies it read, so the caller can compare against them without
+    asking again.
     """
     if not definition.get("origin"):
         raise SyncError(
             f"flow '{target.name}' was not created by sync, so replacing it would "
             "discard whoever drew it. Rename yours, or pass --force."
         )
+    seen: dict[str, str] = {}
     for stored_node in definition.get("nodes") or []:
         node_id = str(stored_node.get("id"))
         try:
-            code = client.get_source(target.name, node_id)
+            entry = client.get_source_entry(target.name, node_id)
         except ApiError:
             continue
+        if entry.get("missing"):
+            continue
+        code = str(entry.get("code", ""))
+        seen[node_id] = code
         if code and not code.startswith(MARKER):
             raise SyncError(
                 f"node '{target.name}.{node_id}' was edited on the canvas, and "
                 "syncing would throw that edit away. Copy it out, or pass --force."
             )
+    return seen
diff --git a/backend/tests/api/routes/test_sync.py b/backend/tests/api/routes/test_sync.py
index fef81bf..8431561 100644
--- a/backend/tests/api/routes/test_sync.py
+++ b/backend/tests/api/routes/test_sync.py
@@ -87,6 +87,31 @@ def test_sync_refuses_to_overwrite_a_canvas_edit(api, flows):
     assert api.get_source("train", "evaluate").startswith(MARKER)
 
 
+def test_an_interrupted_sync_finishes_on_the_next_one(api, flows):
+    """A sync that died between the document and the bodies must be resumable.
+
+    That state — nodes named by a stored flow, with no source written for them
+    yet — used to read as a canvas edit, because the engine answers a new-node
+    template for a node with no body and the template carries no marker. The
+    advice was `--force`, for a canvas nobody had touched.
+    """
+    target = flows["train"]
+    target.name = "resumed"
+    try:
+        # Exactly what the document PUT leaves behind before any body is written.
+        api.put_flow(target.document(ORIGIN) | {"version": 1})
+        assert api.get_source_entry("resumed", "fit")["missing"]
+
+        reports = sync([target], api, origin=ORIGIN)
+
+        assert "fit" in reports[0].changed
+        assert api.get_source("resumed", "fit").startswith(MARKER)
+        assert not api.get_source_entry("resumed", "fit")["missing"]
+        assert not api.get_flow("resumed")["has_draft"]
+    finally:
+        target.name = "train"
+
+
 def test_sync_refuses_a_flow_it_did_not_create(api, flows):
     api.put_flow({"name": "drawn", "version": 1, "nodes": [], "mode": "batch"})
     api.publish("drawn", 1)
diff --git a/backend/tests/flow/test_repo_digest.py b/backend/tests/flow/test_repo_digest.py
new file mode 100644
index 0000000..4db7f37
--- /dev/null
+++ b/backend/tests/flow/test_repo_digest.py
@@ -0,0 +1,91 @@
+"""What the engine hashes to decide a repository's code moved.
+
+The commit cannot answer this: an uncommitted tree stamps `-dirty` for every
+run it ever produces, and the node bodies are import shims that pick up
+whatever is on disk when the worker starts.
+"""
+
+from fluksio.flow import runs
+from fluksio.flow.runs import code_digest, repo_digest
+from fluksio.flow.schemas import FlowDef, FlowOrigin
+
+
+def a_repo(tmp_path, body="x = 1\n"):
+    (tmp_path / "pkg").mkdir()
+    (tmp_path / "pkg" / "train.py").write_text(body)
+    return str(tmp_path)
+
+
+def touch(path, body, mtime):
+    """Write and stamp: two writes inside one clock tick look identical."""
+    path.write_text(body)
+    import os
+
+    os.utime(path, ns=(mtime, mtime))
+
+
+def test_the_same_tree_hashes_the_same(tmp_path):
+    repo = a_repo(tmp_path)
+    assert repo_digest(repo) == repo_digest(repo)
+
+
+def test_editing_a_file_changes_it(tmp_path):
+    repo = a_repo(tmp_path)
+    before = repo_digest(repo)
+    touch(tmp_path / "pkg" / "train.py", "x = 2\n", 2_000_000_000_000_000_000)
+
+    assert repo_digest(repo) != before
+
+
+def test_a_new_file_changes_it(tmp_path):
+    repo = a_repo(tmp_path)
+    before = repo_digest(repo)
+    (tmp_path / "pkg" / "helper.py").write_text("y = 1\n")
+
+    assert repo_digest(repo) != before
+
+
+def test_what_is_not_the_project_is_left_out(tmp_path):
+    """A venv is bigger than the project and is not its code."""
+    repo = a_repo(tmp_path)
+    before = repo_digest(repo)
+
+    (tmp_path / ".venv" / "lib").mkdir(parents=True)
+    (tmp_path / ".venv" / "pyvenv.cfg").write_text("home = /usr\n")
+    (tmp_path / ".venv" / "lib" / "numpy.py").write_text("huge = True\n")
+    (tmp_path / "pkg" / "__pycache__").mkdir()
+    (tmp_path / "pkg" / "__pycache__" / "train.py").write_text("compiled\n")
+    (tmp_path / ".git").mkdir()
+    (tmp_path / ".git" / "hook.py").write_text("hook\n")
+    (tmp_path / "pkg" / "notes.md").write_text("prose\n")
+
+    assert repo_digest(repo) == before
+
+
+def test_a_repository_the_engine_cannot_see_is_empty(tmp_path):
+    """Not an error: a worker elsewhere is a normal deployment."""
+    assert repo_digest(str(tmp_path / "nowhere")) == ""
+    assert repo_digest("") == ""
+
+
+def test_only_a_code_defined_flow_has_one(tmp_path):
+    """A flow drawn on the canvas has no repository, and keys as it always did."""
+    repo = a_repo(tmp_path)
+    declared = FlowDef(
+        name="train", mode="batch", origin=FlowOrigin(repo=repo, commit="abc")
+    )
+
+    assert code_digest(declared) == repo_digest(repo)
+    assert code_digest(FlowDef(name="drawn", mode="batch")) == ""
+
+
+def test_unchanged_files_are_not_read_again(tmp_path, monkeypatch):
+    """The warm path is a stat per file, which is what makes this per-submit."""
+    repo = a_repo(tmp_path)
+    repo_digest(repo)
+
+    def refuse(*_args, **_kwargs):
+        raise AssertionError("an unchanged tree must not be re-read")
+
+    monkeypatch.setattr(runs.Path, "read_bytes", refuse)
+    assert repo_digest(repo)
diff --git a/backend/tests/flow/test_stage_cache_build.py b/backend/tests/flow/test_stage_cache_build.py
index 2d036ce..a54dfc1 100644
--- a/backend/tests/flow/test_stage_cache_build.py
+++ b/backend/tests/flow/test_stage_cache_build.py
@@ -18,7 +18,7 @@ from fluksio.flow.store import FlowStore
 SOURCE = "def process(reading, factor):\n    return {'scaled': reading * factor}\n"
 
 
-def a_flow(**params: object) -> FlowDef:
+def a_flow(provides: list[MessageSpec] | None = None, **params: object) -> FlowDef:
     return FlowDef(
         name="house",
         mode="batch",
@@ -27,7 +27,7 @@ def a_flow(**params: object) -> FlowDef:
                 id="scale",
                 params=dict(params),
                 requires=[MessageSpec(name="reading", dtype=DType.FLOAT)],
-                provides=[MessageSpec(name="scaled", dtype=DType.FLOAT)],
+                provides=provides or [MessageSpec(name="scaled", dtype=DType.FLOAT)],
             )
         ],
     )
@@ -38,13 +38,13 @@ def store(tmp_path: Path) -> FlowStore:
     return FlowStore(tmp_path / "flows")
 
 
-def fingerprint_of(store: FlowStore, flow: FlowDef) -> str:
+def fingerprint_of(store: FlowStore, flow: FlowDef, code_digest: str = "") -> str:
     store.write_flow(flow)
     store.write_node_source(flow.name, "scale", SOURCE)
     pipeline = FlowController(store).build_run_pipeline(
         store.read_flow(flow.name),
         state=MemoryState(),
-        run=RunContext(run_id="r-1"),
+        run=RunContext(run_id="r-1", code_digest=code_digest),
     )
     return pipeline.nodes[0].fingerprint
 
@@ -66,6 +66,37 @@ def test_a_source_change_is_a_different_node(store: FlowStore):
     assert pipeline.nodes[0].fingerprint != first
 
 
+def test_moving_the_repository_is_a_different_node(store: FlowStore):
+    """The stored body of a code-defined node is a shim, and it never moves.
+
+    Editing the function it imports — or anything that function calls into —
+    leaves the shim byte-identical, so without this a re-run is served from
+    the cache and answers with the old code's numbers.
+    """
+    first = fingerprint_of(store, a_flow(factor=3), code_digest="abc")
+
+    assert fingerprint_of(store, a_flow(factor=3), code_digest="def") != first
+    assert fingerprint_of(store, a_flow(factor=3), code_digest="abc") == first
+    # A canvas flow has no repository to hash, and keys the way it always did.
+    assert fingerprint_of(store, a_flow(factor=3)) == fingerprint_of(
+        store, a_flow(factor=3)
+    )
+
+
+def test_declaring_a_new_port_is_a_different_node(store: FlowStore):
+    """A hit restores what the node returned, so it must key on what it returns."""
+    first = fingerprint_of(store, a_flow(factor=3))
+    wider = a_flow(
+        provides=[
+            MessageSpec(name="scaled", dtype=DType.FLOAT),
+            MessageSpec(name="offset", dtype=DType.FLOAT),
+        ],
+        factor=3,
+    )
+
+    assert fingerprint_of(store, wider) != first
+
+
 def test_a_node_that_opted_out_carries_none(store: FlowStore):
     flow = a_flow(factor=3)
     flow.nodes[0].cache = False
diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md
index a2fd80e..0bb0dc2 100644
--- a/docs/concepts/runs.md
+++ b/docs/concepts/runs.md
@@ -177,18 +177,28 @@ downloadable at `GET /api/v1/artifacts/{digest}`.
 ## Stage caching
 
 A run mostly does not redo what an earlier one already did. Before a node
-executes it is fingerprinted — a sha256 over its source, its settings and the
-values it is about to read — and if some earlier run of that same fingerprint
-finished, what that one returned is restored into this run's state and the node
-is skipped. It is recorded with the status `cached` and a duration of zero, and
-its artifacts are listed on the new run as well, so they stay downloadable from
-either.
+executes it is fingerprinted — a sha256 over its source, its settings, the ports
+it declares and the values it is about to read — and if some earlier run of that
+same fingerprint finished, what that one returned is restored into this run's
+state and the node is skipped. It is recorded with the status `cached` and a
+duration of zero, and its artifacts are listed on the new run as well, so they
+stay downloadable from either.
 
 The settings go into the key raw, so a secret contributes its `{"$secret":
 name}` reference and never its value. An artifact input counts as its content
 digest: the same bytes under a different filename are the same input. Each node
 on a run carries the `cache_key` it was looked up by.
 
+For a [code-defined flow](../getting-started/data-science.md), "its source" is
+the generated shim, which imports the real function and does not change when
+that function does. So the key carries one thing more: a digest of every
+`.py` file under the repository the flow was declared in, read when the run
+starts. Editing a helper three calls down from the node invalidates it, which
+is the point — the alternative is a re-run answering with the previous code's
+numbers. It is deliberately blunt: any edit anywhere in the repository re-runs
+every node of its flows. An engine that cannot see the repository — a worker on
+another machine — records no digest and keys as it did before.
+
 The run history *is* the cache; there is no second store. A node's returned
 outputs are kept on its run record as canonical JSON, up to 256000 characters —
 a node returning more than that is simply not cacheable that run. An entry
diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md
index 6f94dfa..876ca13 100644
--- a/docs/getting-started/data-science.md
+++ b/docs/getting-started/data-science.md
@@ -416,12 +416,15 @@ already up, so it is for the run you want now rather than the loop you are
 iterating in.
 
 A second run of a flow you did not change mostly does not execute. Each of
-your nodes is fingerprinted by its source, its settings and the values it
-reads, and one an earlier run already answered is restored from that run
-rather than run again — reported as `cached`, so a flow with nothing left to
-do finishes as `(3/3 cached)`. Change `--lr` and only the nodes downstream of
-it run. `--no-cache` turns that off for one run, `@node(..., cache=False)` for
-one node; the caveat and the details are in
+your nodes is fingerprinted by its settings, the ports it declares, the values
+it reads, and a digest of your repository's Python files — so editing anything
+your node calls into invalidates it, not only the decorated function itself.
+One an earlier run already answered is restored from that run rather than run
+again — reported as `cached`, so a flow with nothing left to do finishes as
+`(3/3 cached)`. Change `--lr` and only the nodes downstream of it run. The
+digest is blunt on purpose: an edit anywhere in the repository re-runs
+everything. `--no-cache` turns caching off for one run, `@node(...,
+cache=False)` for one node; the caveat and the details are in
 [Stage caching](../concepts/runs.md#stage-caching).
 
 From Python, the flow you declared is also the handle to its runs:
@@ -452,6 +455,14 @@ commits — the flow store's and your repository's. That is the answer to "what
 was the learning rate on the run that got 94%?". When a run failed,
 `run.failures` is the node that did it, with its traceback and its logs.
 
+Beside your commit is `code_digest`, and `fluksio runs` prints the pair as
+`a1b2c3d-dirty+9f0e1a2`. The commit alone cannot identify what ran: your node
+bodies are imports, so the engine executes whatever is on disk when the worker
+starts, and an uncommitted tree stamps `-dirty` for every run it ever produces.
+The digest is read at the moment the run starts — so in a sweep whose runs
+queue for hours, each one records the code that actually executed it, not the
+code that was there when you submitted.
+
 ### When the engine is busy
 
 A driver script outlives the engine being slow, because a sweep is hours long