From 91ef2bbe9a427005b265bc3af31d12f84afbd965 Mon Sep 17 00:00:00 2001 From: stroblme Date: Thu, 27 Aug 2026 22:35:23 +0200 Subject: [PATCH] Key a node on the code it reaches, not on the repository around it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `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) Claude-Session: https://claude.ai/code/session_01A9Hdrmf2cwNABCnE5x9UJa --- backend/fluksio/flow/controller.py | 19 +++- backend/fluksio/flow/runs.py | 71 +++++++++++- backend/fluksio/flow/schemas.py | 17 +++ backend/fluksio/sdk/__init__.py | 19 +++- backend/fluksio/sdk/cli.py | 55 +++++++-- backend/fluksio/sdk/client.py | 33 +++++- backend/fluksio/sdk/imports.py | 155 ++++++++++++++++++++++++++ backend/tests/api/routes/test_sync.py | 39 +++++++ backend/tests/sdk/test_imports.py | 101 +++++++++++++++++ docs/code/cli.md | 26 +++-- docs/concepts/runs.md | 21 ++-- docs/getting-started/data-science.md | 10 +- 12 files changed, 530 insertions(+), 36 deletions(-) create mode 100644 backend/fluksio/sdk/imports.py create mode 100644 backend/tests/sdk/test_imports.py diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 06850a0..d883416 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -162,10 +162,15 @@ 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. + #: What this run records as the code it ran: its nodes' digests, hashed. code_digest: str = "" + #: Per node, what the modules that node's function reaches hashed to when + #: the run was claimed. This is what a fingerprint is keyed on, so editing + #: a function the node calls into invalidates the cache the way editing + #: the node itself does — and editing something it does not reach leaves + #: the hit standing. Empty for a node synced before this was recorded, + #: which falls back to `code_digest`. + node_digests: dict[str, str] = field(default_factory=dict) @dataclass(frozen=True, slots=True) @@ -1132,8 +1137,9 @@ class FlowController: # 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 + # covers both: what the modules this node's function + # reaches hash to, which `sync` worked out by following + # its imports. 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( @@ -1149,7 +1155,8 @@ class FlowController: spec.model_dump(mode="json") for spec in node_def.provides ], - "code": run.code_digest, + "code": run.node_digests.get(node_def.id) + or run.code_digest, }, sort_keys=True, separators=(",", ":"), diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 72bf8e6..39f4260 100644 --- a/backend/fluksio/flow/runs.py +++ b/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: diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 8a15757..71acdcb 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -206,6 +206,23 @@ class NodeDef(BaseModel): "workers, which is right for everything that is not compute-heavy." ), ) + code_files: dict[str, str] = Field( + default_factory=dict, + description=( + "The project modules this node's function reaches, dotted name to " + "the file on the machine that synced it. Written by `fluksio " + "sync`, which is the only side that can import the code and see " + "what it imports. Empty for a node drawn on the canvas." + ), + ) + code_digest: str = Field( + default="", + description=( + "What those modules hashed to at sync. The engine hashes them " + "again when a run is claimed, and falls back to this when it " + "cannot see the files — a worker on another machine." + ), + ) @field_validator("id") @classmethod diff --git a/backend/fluksio/sdk/__init__.py b/backend/fluksio/sdk/__init__.py index facf75c..e74e5f6 100644 --- a/backend/fluksio/sdk/__init__.py +++ b/backend/fluksio/sdk/__init__.py @@ -23,6 +23,8 @@ from collections.abc import Callable, Sequence from pathlib import Path from typing import Any, TypeVar +from fluksio.sdk.imports import digest_of, reached + __all__ = [ "FLOWS", "Flow", @@ -549,6 +551,18 @@ def _check(spec: NodeSpec, mode: str) -> dict[str, Any]: } +def _code_of(spec: NodeSpec) -> dict[str, Any]: + """What this node's function calls into, which its shim does not say. + + Read here rather than when the flow is declared, so a second `sync()` in + one process — a notebook — sees the edit that happened between them. This + side is the only one that can work it out at all: it has imported the + code, and the engine never does. + """ + files = reached(spec.fn) + return {"code_files": files, "code_digest": digest_of(files)} + + def _serialisable(value: Any) -> bool: try: json.dumps(value) @@ -676,7 +690,10 @@ class Flow: "name": self.name, "title": self.title, "mode": self.mode, - "nodes": self._nodes, + "nodes": [ + node | _code_of(spec) + for spec, node in zip(self.nodes, self._nodes, strict=True) + ], "inputs": [ {"spec": port.spec(), "initial": port.initial} for port in self.inputs ], diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index e920fc3..a93ca60 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -14,9 +14,10 @@ import importlib import itertools import json import pkgutil +import shutil import sys import time -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterable, Iterator from contextlib import contextmanager, nullcontext from pathlib import Path from typing import Any @@ -744,11 +745,36 @@ def cmd_status(args: argparse.Namespace) -> int: return _unreachable(exc) -#: How much of a run's inputs the list shows. A flow taking a few kB of JSON -#: would otherwise make the table unreadable; `client.runs()` is where the -#: whole value is read. +#: How much of a run's inputs the list shows when nothing says how wide the +#: terminal is. `client.runs()` and `fluksio export runs` are where the whole +#: value is read. PARAMS_WIDTH = 80 +#: What the columns before the inputs take: the id, status, flow, duration and +#: stamp, with their spacing. +LISTING_WIDTH = 84 + + +def _declared(client: Client, flows: Iterable[str]) -> dict[str, dict[str, Any]]: + """What each flow declares its inputs to be, by flow. + + So the listing can leave out an input that is simply the declared value: + what a reader is looking for across a page of runs is where they differ, + and a flow taking a few kB of JSON crowds that off the line. + """ + known: dict[str, dict[str, Any]] = {} + for name in flows: + try: + stored = client.get_flow(name) or {} + except (SyncError, ApiError, httpx.HTTPError): + # A flow deleted since its runs were recorded still lists them. + stored = {} + known[name] = { + str((entry.get("spec") or {}).get("name", "")): entry.get("initial") + for entry in (stored.get("definition") or {}).get("inputs") or [] + } + return known + def _stamp(row: dict[str, Any]) -> str: """What code a run ran: the commit, whether it was dirty, and the digest. @@ -767,14 +793,29 @@ def cmd_runs(args: argparse.Namespace) -> int: try: with _client_for(args, retries=0) as client: rows = client.runs(flow=args.flow, limit=args.limit) + declared = _declared(client, {str(row["flow"]) for row in rows}) except (SyncError, ApiError) as exc: return _fail(str(exc)) except httpx.HTTPError as exc: return _unreachable(exc) + # Whatever is left of the line after the fixed columns; the fallback is + # what a pipe gets, since there is no width to ask for there. + room = max( + 20, + shutil.get_terminal_size((LISTING_WIDTH + PARAMS_WIDTH, 24)).columns + - LISTING_WIDTH, + ) for row in rows: - params = json.dumps(row["params"]) - if len(params) > PARAMS_WIDTH: - params = params[: PARAMS_WIDTH - 3] + "..." + given = row["params"] or {} + defaults = declared.get(str(row["flow"])) or {} + shown = { + name: value + for name, value in given.items() + if name not in defaults or defaults[name] != value + } + params = json.dumps(shown) + if len(params) > room: + params = params[: room - 3] + "..." _say( f"{row['id']} {_status(row['status'], 9)} {row['flow']:<16} " f"{row['duration_ms'] / 1000:7.1f}s {_stamp(row):<22} {params}" diff --git a/backend/fluksio/sdk/client.py b/backend/fluksio/sdk/client.py index 335212b..dad2645 100644 --- a/backend/fluksio/sdk/client.py +++ b/backend/fluksio/sdk/client.py @@ -726,13 +726,19 @@ def _sync_one( if not force: known = _refuse_on_drift(client, target, definition) - saved = client.put_flow(target.document(origin) | {"version": version}) + document = target.document(origin) + moved = _moved_code(stored, document) + saved = client.put_flow(document | {"version": version}) stored_version = int((saved.get("definition") or {}).get("version") or version) if report.created or stored_version != version: # The store bumps the version only when the content actually changed, # so this is its own no-op detection rather than a second guess at it. report.changed.append("flow") version = stored_version + # Named before the shims are compared, because this is the case a shim + # comparison cannot see: the function is unchanged and something it calls + # into is not. + report.changed.extend(moved) for node_id, code in target.shims().items(): if not report.created: @@ -742,7 +748,8 @@ def _sync_one( if stored_code == code: continue client.put_source(target.name, node_id, code) - report.changed.append(node_id) + if node_id not in report.changed: + report.changed.append(node_id) if publish and (client.get_flow(target.name) or {}).get("has_draft"): client.publish(target.name, version) @@ -750,6 +757,28 @@ def _sync_one( return report +def _moved_code(stored: dict[str, Any] | None, document: dict[str, Any]) -> list[str]: + """The nodes whose reachable code changed since the last sync. + + A node is otherwise named only when its generated shim text changes, and + the shim reflects the declared ports rather than what the function calls + into — so editing a helper printed as nothing at all, which reads as + "there was nothing to do". + """ + if stored is None: + return [] + before = { + node.get("id"): node.get("code_digest") or "" + for node in (stored.get("definition") or {}).get("nodes") or [] + } + return [ + node_id + for node in document.get("nodes") or [] + if (node_id := node.get("id")) in before + and before[node_id] != (node.get("code_digest") or "") + ] + + def _refuse_on_drift( client: Client, target: Flow, definition: dict[str, Any] ) -> dict[str, str]: diff --git a/backend/fluksio/sdk/imports.py b/backend/fluksio/sdk/imports.py new file mode 100644 index 0000000..52acdb7 --- /dev/null +++ b/backend/fluksio/sdk/imports.py @@ -0,0 +1,155 @@ +"""What code a node actually reaches, so a cache key can name it. + +A node's stored body is a shim that imports the real function, so it says +nothing about what that function calls into. Hashing the whole repository +instead says too much: it moves when a notebook two directories away does, +which makes the digest neither necessary nor sufficient for "this run's +numbers came from that code". + +This walks the imports out from the module a node's function lives in and +stops at anything installed or standard, so what is left is the project's own +code — including a package installed editable from its own checkout, whose +files are its source rather than a copy under `site-packages`. + +The walk is static: it reads the imports out of each file rather than running +it. A module imported by a name the code computes is therefore not followed, +which is the same limit the yield check has, and for the same reason. +""" + +from __future__ import annotations + +import ast +import hashlib +import importlib.util +import sys +import sysconfig +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +__all__ = ["digest_of", "reached"] + + +#: Never the project's own code, however it happens to be installed. Every +#: node's module imports the authoring API, and from a checkout or an editable +#: install that is a source tree like any other — so without this, a node's +#: digest is mostly this tool, and upgrading it invalidates every cache entry +#: anybody has. +_SELF = frozenset({"fluksio", "fluksio_worker"}) + + +def _foreign() -> tuple[str, ...]: + """Where code that is not the project's lives. + + The standard library and whatever `pip install` copied. An editable + install is deliberately not here: its files are the checkout, and a + checkout is somebody's code. + """ + roots = { + path + for key in ("stdlib", "platstdlib", "purelib", "platlib") + if (path := sysconfig.get_paths().get(key)) + } + roots.update( + entry + for entry in sys.path + if entry.endswith(("site-packages", "dist-packages")) + ) + return tuple(sorted(roots)) + + +def _project_file(name: str, foreign: tuple[str, ...]) -> Path | None: + """The file a dotted module is, if it is the project's own python.""" + module = sys.modules.get(name) + origin = getattr(module, "__file__", None) if module is not None else None + if origin is None: + try: + spec = importlib.util.find_spec(name) + except (ImportError, ValueError, AttributeError): + return None + origin = spec.origin if spec is not None else None + if not origin or not origin.endswith(".py"): + return None + return None if origin.startswith(foreign) else Path(origin) + + +def _package_of(name: str, path: Path) -> str: + """The package a relative import inside this module resolves against.""" + return name if path.name == "__init__.py" else name.rpartition(".")[0] + + +def _imported(tree: ast.AST, package: str) -> Iterator[str]: + """Every module name a file imports, relative ones resolved. + + `from pkg import thing` yields both `pkg` and `pkg.thing`, since only + trying to resolve them says which of the two `thing` is. + """ + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + yield alias.name + elif isinstance(node, ast.ImportFrom): + base = node.module or "" + if node.level: + parts = package.split(".") if package else [] + parts = parts[: len(parts) - (node.level - 1)] + base = ".".join([*parts, base]) if base else ".".join(parts) + if not base: + continue + yield base + for alias in node.names: + yield f"{base}.{alias.name}" + + +def reached(fn: Any) -> dict[str, str]: + """The project modules ``fn``'s own module reaches, itself included. + + Dotted name to the file it is on this machine — the name is what the + digest is keyed on, so an engine hashing the same code arrives at the same + answer; the path is how it finds the file. + """ + start = getattr(fn, "__module__", "") + if not start: + return {} + foreign = _foreign() + found: dict[str, str] = {} + queue = [start] + while queue: + name = queue.pop() + if name in found or name.partition(".")[0] in _SELF: + continue + path = _project_file(name, foreign) + if path is None: + continue + try: + tree = ast.parse(path.read_bytes()) + except (OSError, SyntaxError, ValueError): + continue + found[name] = str(path) + queue.extend(_imported(tree, _package_of(name, path))) + # The packages above it, whose `__init__` runs on the way to importing + # this module and is as much a part of what it does as an import is. + parts = name.split(".")[:-1] + queue.extend(".".join(parts[: index + 1]) for index in range(len(parts))) + return found + + +def digest_of(files: dict[str, str]) -> str: + """What those modules hash to now, or empty if one is not readable here. + + All or nothing: a digest over the half of the code an engine happens to + have would be a confident answer to a question it cannot see. + """ + if not files: + return "" + digest = hashlib.sha256() + for name, path in sorted(files.items()): + try: + body = Path(path).read_bytes() + except OSError: + return "" + digest.update(name.encode()) + digest.update(b"\0") + digest.update(body) + digest.update(b"\0") + return digest.hexdigest() diff --git a/backend/tests/api/routes/test_sync.py b/backend/tests/api/routes/test_sync.py index 8431561..161bb4a 100644 --- a/backend/tests/api/routes/test_sync.py +++ b/backend/tests/api/routes/test_sync.py @@ -65,6 +65,45 @@ def test_sync_stores_a_runnable_flow_with_its_origin(api, flows): assert "from myresearch.train import fit" in source +def test_a_node_carries_the_modules_its_code_reaches(api, flows): + """The shim names one function; the cache has to key on what it calls.""" + sync([flows["train"]], api, origin=ORIGIN) + + nodes = {n["id"]: n for n in api.get_flow("train")["definition"]["nodes"]} + files = nodes["fit"]["code_files"] + + # The module the function is in, and the package whose `__init__` runs on + # the way to it. Not the engine's own code, and not the standard library. + assert "myresearch.train" in files + assert all(path.endswith(".py") for path in files.values()) + assert not any(name.startswith("fluksio") for name in files) + assert len(nodes["fit"]["code_digest"]) == 64 + + +def test_a_node_whose_helper_moved_is_named_by_sync(): + """Editing a helper changes no shim, so nothing used to say it happened.""" + from fluksio.sdk.client import _moved_code + + stored = { + "definition": { + "nodes": [ + {"id": "fit", "code_digest": "aaa"}, + {"id": "prepare", "code_digest": "bbb"}, + ] + } + } + document = { + "nodes": [ + {"id": "fit", "code_digest": "ccc"}, + {"id": "prepare", "code_digest": "bbb"}, + ] + } + + assert _moved_code(stored, document) == ["fit"] + # Nothing to compare against is not a change. + assert _moved_code(None, document) == [] + + def test_a_second_sync_changes_nothing_and_commits_nothing(api, flows): sync([flows["train"]], api, origin=ORIGIN) before = commits() diff --git a/backend/tests/sdk/test_imports.py b/backend/tests/sdk/test_imports.py new file mode 100644 index 0000000..a21e282 --- /dev/null +++ b/backend/tests/sdk/test_imports.py @@ -0,0 +1,101 @@ +"""What a node's code reaches — the narrow answer a cache key is keyed on.""" + +import sys +import textwrap + +from fluksio.sdk.imports import digest_of, reached + + +def _write(root, name, body): + path = root / f"{name}.py" + path.write_text(textwrap.dedent(body).lstrip()) + return path + + +def _study(tmp_path): + """A package whose node function delegates to a helper beside it.""" + root = tmp_path / "study" + root.mkdir() + (root / "__init__.py").write_text("VERSION = 1\n") + _write( + root, + "train", + """ + import json + + from study.helpers import curve + + + def fit(lr): + return json.dumps(curve(lr)) + """, + ) + _write(root, "helpers", "def curve(lr):\n return [lr]\n") + _write(root, "unused", "def nothing():\n return 0\n") + return root + + +def test_the_walk_follows_a_helper_and_stops_at_installed_code(tmp_path, monkeypatch): + """The shim says nothing about this, and the repository says too much.""" + _study(tmp_path) + monkeypatch.syspath_prepend(str(tmp_path)) + for name in [n for n in sys.modules if n == "study" or n.startswith("study.")]: + del sys.modules[name] + + from study.train import fit + + files = reached(fit) + + # The module, the helper it calls into, and the package whose `__init__` + # runs on the way in. Not the sibling nobody imports, and not `json`. + assert set(files) == {"study", "study.train", "study.helpers"} + assert all(path.endswith(".py") for path in files.values()) + + +def test_the_digest_moves_with_the_helper_and_not_with_the_neighbour( + tmp_path, monkeypatch +): + """The whole point: it is neither necessary nor sufficient otherwise.""" + root = _study(tmp_path) + monkeypatch.syspath_prepend(str(tmp_path)) + for name in [n for n in sys.modules if n == "study" or n.startswith("study.")]: + del sys.modules[name] + + from study.train import fit + + files = reached(fit) + before = digest_of(files) + + _write(root, "unused", "def nothing():\n return 1\n") + assert digest_of(files) == before + + _write(root, "helpers", "def curve(lr):\n return [lr, lr]\n") + assert digest_of(files) != before + + # A file this engine cannot see is not half an answer. + assert digest_of({**files, "study.gone": str(root / "gone.py")}) == "" + + +def test_the_tool_itself_is_not_the_projects_code(tmp_path, monkeypatch): + """Every node imports the authoring API, and from a checkout that is source.""" + root = tmp_path / "arm" + root.mkdir() + (root / "__init__.py").write_text("") + _write( + root, + "work", + """ + from fluksio import node + + + def fit(lr): + return lr + """, + ) + monkeypatch.syspath_prepend(str(tmp_path)) + for name in [n for n in sys.modules if n == "arm" or n.startswith("arm.")]: + del sys.modules[name] + + from arm.work import fit + + assert set(reached(fit)) == {"arm", "arm.work"} diff --git a/docs/code/cli.md b/docs/code/cli.md index c954665..46080ea 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -160,7 +160,14 @@ file path, because the shim has to import the same way. Every sync retires the engine's workers, including one that had nothing to upload — a worker holds your package in memory, so an edit to it is invisible -until the process goes. See +until the process goes. + +It also records, per node, which of your modules that node's function imports +its way to, and what they hash to. That is what the +[stage cache](../concepts/runs.md#stage-caching) keys on, so editing a +helper a node calls into is reported as that node changing — `train: updated +(flow, fit)` — and re-runs it, while editing something the node never reaches +is left alone. See [Getting started: data science](../getting-started/data-science.md). ### `fluksio run` @@ -275,12 +282,17 @@ fluksio runs [--flow train] [--limit 20] ``` The runs an engine has recorded, newest first: id, status, flow, duration, the -commit of the repository it came from, and its parameters. Statuses are -coloured when a terminal is reading the output — `ok` green, `error` red, -`cached` cyan. Parameters are clamped to 80 characters so a flow taking a few -kB of JSON still lists as a table; `Client.runs()` is where the whole value is -read. `--local` reads the same history from an in-process engine, without one -having to be served. +commit of the repository it came from, and the inputs it was given. Statuses +are coloured when a terminal is reading the output — `ok` green, `error` red, +`cached` cyan. + +Only the inputs that *differ from what the flow declares* are shown, and they +are clamped to what is left of the terminal's width — a run that took the +defaults lists none at all, and a flow taking a few kB of JSON does not push +everything else off the line. `Client.runs()` and +[`fluksio export runs`](#fluksio-export) are where the whole value is read. +`--local` reads the same history from an in-process engine, without one having +to be served. ### `fluksio flavors` diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 0cfa433..bbafd5f 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -191,13 +191,20 @@ 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. +that function does. So the key carries one thing more: a digest of the project +modules that node's function reaches, worked out by `fluksio sync` — which is +the only side that imports your code and can see what it imports — and read +again from those files 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. +Editing something the node does not reach leaves the hit standing, which is +the other half — a notebook two directories away is not a reason to retrain. +The walk follows imports statically and stops at the standard library, at +anything installed, and at Fluksio itself; a module imported under a name the +code computes is not followed. An engine that cannot see the files keeps what +sync recorded instead of nothing, so a worker on another machine no longer +keys every run the same. 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 — diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index 4bf66e8..d156971 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -542,9 +542,13 @@ 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. +The digest is over the modules this flow's nodes actually reach — not the +whole repository, so it moves when the code behind a number moves and stays +put when a notebook beside it changes — and it 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 rather than the code that was there when you +submitted. It is what makes the column worth joining an +[exported table](../concepts/runs.md#taking-it-into-a-dataframe) on. ### When the engine is busy