refactor: unify environment hashing on environment_content_hash

src/harbor/environments/blaxel.py

@@ -24,9 +24,9 @@
 )
 from harbor.environments.dind_compose import DinDComposeOps
 from harbor.environments.definition import (
+    SNAPSHOT_HASH_LEN,
     COMPOSE_FILE_NAME,
     effective_exec_cwd,
-    environment_template_hash,
     parse_dockerfile_workdir,
     require_agent_environment_definition,
     should_use_prebuilt_docker_image,
@@ -200,18 +200,14 @@ def __init__(
             )
             or "/"
         )
-        env_hash = environment_template_hash(
-            self.environment_dir,
-            docker_image=task_env_config.docker_image,
-            environment_name=environment_name,
-        )
+        env_hash = self.environment_id[:SNAPSHOT_HASH_LEN]
         # Image identity covers the task content and the provider's build
         # config: registered images are immutable per name, so a build-config
         # change must produce a new name rather than collide with artifacts
         # built under the old config.
         build_hash = hashlib.sha256(
             f"{env_hash}:{_NO_SLIM_BLAXEL_TOML}".encode()
-        ).hexdigest()[:8]
+        ).hexdigest()[:SNAPSHOT_HASH_LEN]
         self._image_name = _sanitize_blaxel_name(
             f"harbor-img-{environment_name}-{build_hash}"
         )

src/harbor/environments/daytona/environment.py

@@ -36,6 +36,7 @@
 )
 from harbor.environments.dind_compose import DinDComposeOps
 from harbor.environments.definition import (
+    SNAPSHOT_HASH_LEN,
     require_agent_environment_definition,
     should_use_prebuilt_docker_image,
 )
@@ -1193,17 +1194,11 @@ def _snapshots(self) -> DaytonaSnapshotService:
         if self._snapshot_service is None:
             self._snapshot_service = DaytonaSnapshotService(
                 logger=self.logger,
-                environment_dir=self.environment_dir,
+                env_hash=self.environment_id[:SNAPSHOT_HASH_LEN],
                 dockerfile_path=self._dockerfile_path,
             )
         return self._snapshot_service
 
-    def _get_environment_hash(self) -> str:
-        return self._snapshots().environment_hash()
-
-    def _get_auto_snapshot_name(self) -> str:
-        return self._snapshots().auto_snapshot_name()
-
     def _sandbox_common_kwargs(self) -> dict[str, Any]:
         return {
             "auto_delete_interval": self._auto_delete_interval,

src/harbor/environments/daytona/snapshots.py

@@ -18,7 +18,6 @@
     SNAPSHOT_GET_RETRY,
     SNAPSHOT_GET_WAIT,
 )
-from harbor.utils.container_cache import environment_dir_hash_truncated
 
 if TYPE_CHECKING:
     from daytona import AsyncDaytona, CreateSnapshotParams, Image, Resources
@@ -76,22 +75,18 @@ def __init__(
         self,
         *,
         logger: Logger,
-        environment_dir: Path,
+        env_hash: str,
         dockerfile_path: Path,
     ) -> None:
         self._logger = logger
-        self._environment_dir = environment_dir
+        self._env_hash = env_hash
         self._dockerfile_path = dockerfile_path
 
-    def environment_hash(self, *, truncate: int = 12) -> str:
-        return environment_dir_hash_truncated(self._environment_dir, truncate=truncate)
-
     def auto_snapshot_name(self) -> str:
-        env_hash = self.environment_hash()
         target = os.environ.get("DAYTONA_TARGET")
         if target:
-            return f"harbor__{env_hash}__{target}__snapshot"
-        return f"harbor__{env_hash}__snapshot"
+            return f"harbor__{self._env_hash}__{target}__snapshot"
+        return f"harbor__{self._env_hash}__snapshot"
 
     async def resolve_template(
         self,

src/harbor/environments/definition.py

@@ -72,22 +72,11 @@ def require_agent_environment_definition(
     )
 
 
-def environment_template_hash(
-    environment_dir: Path,
-    *,
-    docker_image: str | None = None,
-    environment_name: str,
-) -> str:
-    if any(environment_dir.iterdir()):
-        from dirhash import dirhash
-
-        return dirhash(environment_dir, "sha256")[:8]
-    seed = docker_image or environment_name
-    return hashlib.sha256(seed.encode()).hexdigest()[:8]
-
-
 _CONTENT_HASH_IGNORE_NAMES = frozenset({".DS_Store", ".git", "__pycache__"})
 
+# Length of hash suffixes embedded in provider snapshot/template/image names.
+SNAPSHOT_HASH_LEN = 12
+
 
 def environment_content_hash(
     environment_dir: Path,

src/harbor/environments/e2b.py

@@ -17,8 +17,8 @@
     EnvironmentResourceCapabilities,
 )
 from harbor.environments.definition import (
+    SNAPSHOT_HASH_LEN,
     effective_exec_cwd,
-    environment_template_hash,
     parse_dockerfile_workdir,
     require_agent_environment_definition,
 )
@@ -102,11 +102,7 @@ def __init__(
         self._workdir = parse_dockerfile_workdir(self._environment_definition_path)
 
         self._sandbox: AsyncSandbox | None = None
-        env_hash = environment_template_hash(
-            self.environment_dir,
-            docker_image=task_env_config.docker_image,
-            environment_name=environment_name,
-        )
+        env_hash = self.environment_id[:SNAPSHOT_HASH_LEN]
         self._template_name = f"{environment_name}__{env_hash}".replace(
             "/", "__"
         ).replace(".", "-")

src/harbor/environments/langsmith.py

@@ -23,7 +23,10 @@
     ServiceOperationsUnsupportedError,
 )
 from harbor.environments.capabilities import EnvironmentCapabilities
-from harbor.environments.definition import should_use_prebuilt_docker_image
+from harbor.environments.definition import (
+    SNAPSHOT_HASH_LEN,
+    should_use_prebuilt_docker_image,
+)
 from harbor.environments.docker import (
     COMPOSE_BUILD_PATH,
     COMPOSE_NO_NETWORK_PATH,
@@ -42,7 +45,6 @@
 from harbor.models.task.config import EnvironmentConfig
 from harbor.models.trial.config import ResourceMode, ServiceVolumeConfig
 from harbor.models.trial.paths import EnvironmentPaths, TrialPaths
-from harbor.utils.container_cache import environment_dir_hash_truncated
 from harbor.utils.env import resolve_env_vars
 from harbor.utils.optional_import import MissingExtraError
 
@@ -948,7 +950,7 @@ async def _resolve_snapshot_name(self, force_build: bool) -> str:
     async def _resolve_dockerfile_snapshot_name(
         self, force_build: bool, dockerfile: Path
     ) -> str:
-        env_hash = environment_dir_hash_truncated(self.environment_dir)
+        env_hash = self.environment_id[:SNAPSHOT_HASH_LEN]
         fs_capacity_bytes = self._fs_capacity_bytes(
             minimum_mb=_DEFAULT_SNAPSHOT_STORAGE_MB
         )

src/harbor/environments/novita.py

@@ -48,7 +48,7 @@
     EnvironmentResourceCapabilities,
 )
 from harbor.environments.definition import (
-    environment_template_hash,
+    SNAPSHOT_HASH_LEN,
     parse_dockerfile_workdir,
     require_agent_environment_definition,
 )
@@ -761,11 +761,7 @@ def __init__(
             )
 
         key_suffix = self._api_key[-4:].lower()
-        env_hash = environment_template_hash(
-            self.environment_dir,
-            docker_image=task_env_config.docker_image,
-            environment_name=environment_name,
-        )
+        env_hash = self.environment_id[:SNAPSHOT_HASH_LEN]
         self._template_name = (
             f"{environment_name}__{env_hash}_{key_suffix}".replace("/", "__")
             .replace(".", "-")

src/harbor/models/task/task.py

@@ -1,5 +1,6 @@
 import re
 import tomllib
+import warnings
 from pathlib import Path
 
 from pydantic import ValidationError
@@ -195,7 +196,17 @@ def step_instruction(self, step_name: str) -> str:
 
     @property
     def checksum(self) -> str:
-        """Generate a deterministic hash for the task based on its entire directory content."""
+        """Generate a deterministic hash for the task based on its entire directory content.
+
+        This property will be deprecated soon. Use ``TrialLock.task.digest`` from
+        the trial lock when you need the durable content hash for a task.
+        """
+        warnings.warn(
+            "Task.checksum will be deprecated soon. Use TrialLock.task.digest "
+            "from the trial lock when you need the task content hash.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         from dirhash import dirhash
 
         return dirhash(self._task_dir, "sha256")

src/harbor/utils/container_cache.py

@@ -7,41 +7,6 @@
 from stat import S_ISLNK, S_ISREG
 
 
-def dockerfile_hash(path: Path) -> str:
-    """Full SHA256 of a Dockerfile for cache lookups."""
-    return hashlib.sha256(path.read_bytes()).hexdigest()
-
-
-def dockerfile_hash_truncated(path: Path, truncate: int = 12) -> str:
-    """Truncated SHA256 hash of Dockerfile content."""
-    return dockerfile_hash(path)[:truncate]
-
-
-def environment_dir_hash(env_dir: Path) -> str:
-    """Full SHA256 of the entire environment directory.
-
-    Hashes every file's relative path and content so that two environment
-    directories with the same Dockerfile but different fixture files produce
-    different hashes. Files are processed in sorted order for determinism.
-    """
-    h = hashlib.sha256()
-    for file_path in sorted(env_dir.rglob("*")):
-        if file_path.is_file():
-            rel = str(file_path.relative_to(env_dir))
-            rel_bytes = rel.encode("utf-8")
-            h.update(len(rel_bytes).to_bytes(4, "big"))
-            h.update(rel_bytes)
-            content = file_path.read_bytes()
-            h.update(len(content).to_bytes(4, "big"))
-            h.update(content)
-    return h.hexdigest()
-
-
-def environment_dir_hash_truncated(env_dir: Path, truncate: int = 12) -> str:
-    """Truncated SHA256 hash of the entire environment directory."""
-    return environment_dir_hash(env_dir)[:truncate]
-
-
 def docker_build_context_hash(
     *,
     context: Path,

tests/unit/environments/test_daytona_snapshots.py

@@ -1,4 +1,4 @@
-"""Unit tests for DaytonaSnapshotService naming and hashing."""
+"""Unit tests for DaytonaSnapshotService naming."""
 
 from pathlib import Path
 
@@ -9,14 +9,11 @@
 
 
 @pytest.fixture
-def snapshot_service(temp_dir: Path) -> DaytonaSnapshotService:
-    env_dir = temp_dir / "environment"
-    env_dir.mkdir()
-    (env_dir / "Dockerfile").write_text("FROM ubuntu:22.04\n")
+def snapshot_service() -> DaytonaSnapshotService:
     return DaytonaSnapshotService(
         logger=logger.getChild("test"),
-        environment_dir=env_dir,
-        dockerfile_path=env_dir / "Dockerfile",
+        env_hash="abc123def456",
+        dockerfile_path=Path("/task/environment/Dockerfile"),
     )
 
 
@@ -25,14 +22,13 @@ def test_snapshot_name_without_target(
         self, snapshot_service: DaytonaSnapshotService, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         monkeypatch.delenv("DAYTONA_TARGET", raising=False)
-        env_hash = snapshot_service.environment_hash()
-        assert snapshot_service.auto_snapshot_name() == f"harbor__{env_hash}__snapshot"
+        assert snapshot_service.auto_snapshot_name() == "harbor__abc123def456__snapshot"
 
     def test_snapshot_name_includes_daytona_target(
         self, snapshot_service: DaytonaSnapshotService, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         monkeypatch.setenv("DAYTONA_TARGET", "RL")
-        env_hash = snapshot_service.environment_hash()
         assert (
-            snapshot_service.auto_snapshot_name() == f"harbor__{env_hash}__RL__snapshot"
+            snapshot_service.auto_snapshot_name()
+            == "harbor__abc123def456__RL__snapshot"
         )

tests/unit/environments/test_environment_definition.py

@@ -9,7 +9,7 @@
 from harbor.environments.daytona import DaytonaEnvironment, _DaytonaDirect
 from harbor.environments.definition import (
     effective_exec_cwd,
-    environment_template_hash,
+    environment_content_hash,
     has_agent_environment_definition,
     parse_dockerfile_workdir,
     require_agent_environment_definition,
@@ -52,18 +52,6 @@ def test_require_agent_environment_definition_raises(self, temp_dir):
         with pytest.raises(FileNotFoundError, match="no environment definition"):
             require_agent_environment_definition(_empty_env_dir(temp_dir))
 
-    def test_environment_template_hash_uses_image_when_dir_empty(self, temp_dir):
-        env_dir = _empty_env_dir(temp_dir)
-
-        h1 = environment_template_hash(
-            env_dir, docker_image="ubuntu:22.04", environment_name="task-a"
-        )
-        h2 = environment_template_hash(
-            env_dir, docker_image="ubuntu:22.04", environment_name="task-b"
-        )
-        assert h1 == h2
-        assert len(h1) == 8
-
     def test_parse_dockerfile_workdir(self, temp_dir):
         dockerfile = temp_dir / "Dockerfile"
         assert parse_dockerfile_workdir(dockerfile) is None
@@ -140,6 +128,65 @@ def test_should_upload_environment_dir(self, temp_dir):
         assert not should_upload_environment_dir(env_dir, docker_image="ubuntu:22.04")
 
 
+class TestEnvironmentContentHash:
+    def test_deterministic(self, temp_dir):
+        env_dir = _empty_env_dir(temp_dir)
+        (env_dir / "Dockerfile").write_text("FROM ubuntu:22.04\n")
+        (env_dir / "setup.sh").write_text("#!/bin/sh\necho ok\n")
+
+        assert environment_content_hash(env_dir) == environment_content_hash(env_dir)
+
+    def test_fixture_change_changes_hash(self, temp_dir):
+        env_dir = _empty_env_dir(temp_dir)
+        (env_dir / "Dockerfile").write_text("FROM ubuntu:22.04\n")
+        (env_dir / "fixture.txt").write_text("v1")
+        hash_v1 = environment_content_hash(env_dir)
+
+        (env_dir / "fixture.txt").write_text("v2")
+        assert environment_content_hash(env_dir) != hash_v1
+
+    def test_ignores_transient_files(self, temp_dir):
+        env_dir = _empty_env_dir(temp_dir)
+        (env_dir / "Dockerfile").write_text("FROM ubuntu:22.04\n")
+        before = environment_content_hash(env_dir)
+
+        (env_dir / ".DS_Store").write_bytes(b"junk")
+        pycache = env_dir / "__pycache__"
+        pycache.mkdir()
+        (pycache / "mod.cpython-312.pyc").write_bytes(b"junk")
+
+        assert environment_content_hash(env_dir) == before
+
+    def test_path_and_content_boundary_do_not_collide(self, temp_dir):
+        """File 'ab' (empty) must not hash the same as file 'a' with content 'b'."""
+        left = temp_dir / "left"
+        right = temp_dir / "right"
+        left.mkdir()
+        right.mkdir()
+        (left / "ab").write_bytes(b"")
+        (right / "a").write_bytes(b"b")
+
+        assert environment_content_hash(left) != environment_content_hash(right)
+
+    def test_truncated_hash_is_prefix(self, temp_dir):
+        env_dir = _empty_env_dir(temp_dir)
+        (env_dir / "Dockerfile").write_text("FROM alpine:3.19\n")
+
+        truncated = environment_content_hash(env_dir, truncate=12)
+        assert environment_content_hash(env_dir).startswith(truncated)
+        assert len(truncated) == 12
+
+    def test_uses_image_seed_when_dir_empty(self, temp_dir):
+        env_dir = _empty_env_dir(temp_dir)
+
+        h1 = environment_content_hash(env_dir, docker_image="ubuntu:22.04", truncate=12)
+        h2 = environment_content_hash(env_dir, docker_image="ubuntu:22.04", truncate=12)
+        h3 = environment_content_hash(env_dir, docker_image="ubuntu:24.04", truncate=12)
+        assert h1 == h2
+        assert h1 != h3
+        assert len(h1) == 12
+
+
 class TestProviderValidation:
     def test_docker_accepts_docker_image_without_dockerfile(self, temp_dir):
         with patch.object(