diff --git a/src/horus_builtin/runtime/python_script.py b/src/horus_builtin/runtime/python_script.py index 11402e2..4813376 100644 --- a/src/horus_builtin/runtime/python_script.py +++ b/src/horus_builtin/runtime/python_script.py @@ -81,6 +81,15 @@ def anchor_local_paths(self, base: Path) -> None: if not self.script.is_absolute(): self.script = (base / self.script).resolve() + def local_files(self) -> list[Path]: + """ + The script, unless it is templated: a templated script names an + input artifact, already digested through the task's inputs. + """ + if _is_template(self.script): + return [] + return [self.script] + async def _setup_runtime(self, task: "BaseTask") -> str: if _is_template(self.script): # ``script: ${my_script}`` names an input artifact instead of a diff --git a/src/horus_builtin/task/horus_task.py b/src/horus_builtin/task/horus_task.py index 34e2b70..a8ffafe 100644 --- a/src/horus_builtin/task/horus_task.py +++ b/src/horus_builtin/task/horus_task.py @@ -21,8 +21,8 @@ import hashlib import json -from pathlib import PurePosixPath -from typing import ClassVar +from pathlib import Path, PurePosixPath +from typing import Any, ClassVar from pydantic import BaseModel @@ -37,6 +37,19 @@ from horus_runtime.i18n import tr as _ _UNHASHABLE = "unhashable" + + +def _digest_file(path: Path) -> str: + """ + sha256 of a local file, read in chunks. + """ + sha = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(1024 * 1024), b""): + sha.update(block) + return sha.hexdigest() + + """Fingerprint value for an input the target cannot digest.""" @@ -136,16 +149,39 @@ async def _fingerprint(self) -> TaskFingerprint: artifact.id: await store.digest(artifact) or _UNHASHABLE for artifact in self.inputs } - config = json.dumps( - { - "runtime": self.runtime.model_dump(mode="json"), - "executor": self.executor.model_dump(mode="json"), - }, - sort_keys=True, - ) + payload: dict[str, Any] = { + "runtime": self.runtime.model_dump(mode="json"), + "executor": self.executor.model_dump(mode="json"), + } + # A runtime holds its script as a path, so the dump above changes + # when the path changes but not when the file does. Without this a + # task keeps skipping after its code was edited. + code = self._local_file_digests() + if code: + payload["code"] = code + config = json.dumps(payload, sort_keys=True) config_hash = hashlib.sha256(config.encode()).hexdigest() return TaskFingerprint(inputs=inputs, config_hash=config_hash) + def _local_file_digests(self) -> list[list[str]]: + """ + Sorted ``[name, sha256]`` for the runtime's and executor's local + files, so the same set hashes the same way twice. + + Keyed by name, not path, to add no new path dependence. A missing + file is skipped: it drops out of the set, so the task re-runs. + """ + digests: list[list[str]] = [] + for path in [ + *self.runtime.local_files(), + *self.executor.local_files(), + ]: + try: + digests.append([path.name, _digest_file(path)]) + except OSError: + continue + return sorted(digests) + async def _write_manifest(self, fingerprint: TaskFingerprint) -> None: """ Record *fingerprint* on the target, next to the outputs the run just diff --git a/src/horus_runtime/core/executor/base.py b/src/horus_runtime/core/executor/base.py index 99de8fb..9fe5bbc 100644 --- a/src/horus_runtime/core/executor/base.py +++ b/src/horus_runtime/core/executor/base.py @@ -133,6 +133,16 @@ def anchor_local_paths(self, base: Path) -> None: working directory instead. """ + def local_files(self) -> list[Path]: + """ + Local files this executor reads from the orchestrator, digested + into the task fingerprint. Empty by default. + + Mirrors :meth:`BaseRuntime.local_files`; a conda + ``environment_file`` is the case this exists for. + """ + return [] + @final async def execute(self, task: "BaseTask") -> None: """ diff --git a/src/horus_runtime/core/runtime/base.py b/src/horus_runtime/core/runtime/base.py index 8f1ce80..d079eaa 100644 --- a/src/horus_runtime/core/runtime/base.py +++ b/src/horus_runtime/core/runtime/base.py @@ -65,6 +65,16 @@ def anchor_local_paths(self, base: Path) -> None: Called by the workflow before execution. No-op by default. """ + def local_files(self) -> list[Path]: + """ + Local files this runtime reads from the orchestrator, digested + into the task fingerprint. Empty by default. + + Mirrors :meth:`anchor_local_paths`: whatever that resolves is + what this returns. + """ + return [] + @abstractmethod async def _setup_runtime(self, task: "BaseTask") -> T: """ diff --git a/tests/unit/task/test_builtin_task.py b/tests/unit/task/test_builtin_task.py index d3ca252..0829abd 100644 --- a/tests/unit/task/test_builtin_task.py +++ b/tests/unit/task/test_builtin_task.py @@ -20,6 +20,8 @@ Unit tests for HorusTask builtin task. """ +import hashlib +import json from pathlib import Path from unittest.mock import patch @@ -29,6 +31,7 @@ from horus_builtin.artifact.file import FileArtifact from horus_builtin.executor.shell import ShellExecutor from horus_builtin.runtime.command import CommandRuntime +from horus_builtin.runtime.python_script import PythonScriptRuntime from horus_builtin.target.local import LocalTarget from horus_builtin.task.horus_task import HorusTask from horus_runtime.core.artifact.exceptions import ArtifactDoesNotExistError @@ -457,3 +460,125 @@ async def test_does_not_skip_without_a_manifest( assert task.runs == 1 assert await task.is_complete() is True + + +@pytest.mark.unit +class TestLocalFilesInFingerprint: + """ + A runtime holds its script as a path, so the config dump changes when + the path changes but not when the file does. Without the file's digest + a task keeps skipping after its code was edited. + """ + + @staticmethod + def _script_task(tmp_path: Path, body: str) -> HorusTask: + """ + A task whose runtime runs a real script file on disk. + """ + script = tmp_path / "prep.py" + script.write_text(body) + return HorusTask( + id="prep", + name="Prepare", + runtime=PythonScriptRuntime(script=script), + executor=ShellExecutor(), + target=LocalTarget(working_directory=str(tmp_path)), + ) + + async def test_editing_a_script_changes_the_fingerprint( + self, tmp_path: Path + ) -> None: + """ + The bug this exists to fix: the task used to skip with stale code. + """ + task = self._script_task(tmp_path, "print('one')\n") + before = (await task._fingerprint()).config_hash + + (tmp_path / "prep.py").write_text("print('two')\n") + after = (await task._fingerprint()).config_hash + + assert before != after + + async def test_an_untouched_script_keeps_its_fingerprint( + self, tmp_path: Path + ) -> None: + """ + Hashing the file must not make the fingerprint unstable, or every + run would re-run everything. + """ + task = self._script_task(tmp_path, "print('one')\n") + assert (await task._fingerprint()).config_hash == ( + await task._fingerprint() + ).config_hash + + async def test_a_task_without_local_files_is_unaffected( + self, tmp_path: Path, make_shell_task: MakeTaskType + ) -> None: + """ + The code entry is only added when a task has local files, so + existing caches for command-only tasks stay valid. + """ + del tmp_path + task = make_shell_task(cmd="echo hello") + fingerprint = await task._fingerprint() + payload = json.dumps( + { + "runtime": task.runtime.model_dump(mode="json"), + "executor": task.executor.model_dump(mode="json"), + }, + sort_keys=True, + ) + assert ( + fingerprint.config_hash + == hashlib.sha256(payload.encode()).hexdigest() + ) + + async def test_a_missing_script_is_skipped_rather_than_raising( + self, tmp_path: Path + ) -> None: + """ + It drops out of the set, which changes the hash, which re-runs the + task. A missing script deserves exactly that. + """ + task = self._script_task(tmp_path, "print('one')\n") + with_script = (await task._fingerprint()).config_hash + + (tmp_path / "prep.py").unlink() + without = (await task._fingerprint()).config_hash + + assert with_script != without + + +@pytest.mark.unit +class TestLocalFilesHooks: + """ + What each base class reports as its own local files. + """ + + def test_a_runtime_owns_nothing_by_default(self) -> None: + """ + A runtime carrying its command inline reads no local file. + """ + assert CommandRuntime(command="echo hi").local_files() == [] + + def test_an_executor_owns_nothing_by_default(self) -> None: + """ + The hook exists for executors that point at an environment file. + """ + assert ShellExecutor().local_files() == [] + + def test_a_script_runtime_owns_its_script(self, tmp_path: Path) -> None: + """ + The case the fingerprint change depends on. + """ + script = tmp_path / "prep.py" + script.write_text("print('one')\n") + assert PythonScriptRuntime(script=script).local_files() == [script] + + def test_a_templated_script_owns_nothing(self) -> None: + """ + A templated script names an input artifact, whose digest is + already in the fingerprint through the task's inputs. + """ + runtime = PythonScriptRuntime(script=Path("${my_script}")) + assert runtime.local_files() == []