From bf19c5e5701fd6fd8805f12eac75e0ccb1f44a8b Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:08:04 +0200
Subject: [PATCH 01/18] fix: hash objects by __slots__/__dict__ state instead
 of memory address

Objects without __dict__ (e.g. @dataclass(slots=True), __slots__ value
types) fell through to repr(obj), embedding the memory address. Identical
calls produced a new cache key every instantiation/process (permanent
misses), and lossy custom reprs could collide (wrong cached result).

_stable_repr_to now reconstructs object state from both __dict__ and the
MRO's __slots__ and hashes by value. Genuinely opaque objects (no state,
default object.__repr__) now emit UnhashableArgWarning instead of silently
address-hashing.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/hashing.py | 68 ++++++++++++++++++++++++++++++++-------
 tests/test_hashing.py | 74 ++++++++++++++++++++++++++++++++++++++++++-
 2 files changed, 130 insertions(+), 12 deletions(-)

diff --git a/src/cashet/hashing.py b/src/cashet/hashing.py
index 5cfa771..63c2ebf 100644
--- a/src/cashet/hashing.py
+++ b/src/cashet/hashing.py
@@ -21,6 +21,10 @@ class ClosureWarning(UserWarning):
     pass
 
 
+class UnhashableArgWarning(UserWarning):
+    pass
+
+
 _pickle_warning_issued = False
 
 
@@ -376,6 +380,36 @@ def hash_function(
     return h.hexdigest()
 
 
+def _slot_names(cls: type) -> list[str]:
+    names: list[str] = []
+    for klass in cls.__mro__:
+        slots = klass.__dict__.get("__slots__")
+        if slots is None:
+            continue
+        if isinstance(slots, str):
+            slots = (slots,)
+        for name in slots:
+            if name in ("__dict__", "__weakref__"):
+                continue
+            names.append(name)
+    return names
+
+
+_UNSET = object()
+
+
+def _object_state(obj: Any) -> dict[str, Any] | None:
+    state: dict[str, Any] = {}
+    instance_dict = getattr(obj, "__dict__", None)
+    if isinstance(instance_dict, dict):
+        state.update(instance_dict)
+    for name in _slot_names(type(obj)):
+        value = getattr(obj, name, _UNSET)
+        if value is not _UNSET:
+            state[name] = value
+    return state or None
+
+
 def _stable_repr_to(
     buf: io.StringIO, obj: Any, _visited: set[int] | None = None
 ) -> None:
@@ -453,18 +487,30 @@ def _stable_repr_to(
         buf.write(f"<type:{obj.__module__}.{obj.__qualname__}>")
     elif hasattr(obj, "__cashet_ref__"):
         buf.write(f"<ref:{obj.__cashet_ref__().hash}>")
-    elif hasattr(obj, "__dict__"):
-        obj_id = id(obj)
-        if obj_id in _visited:
-            buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:...>")
-            return
-        _visited.add(obj_id)
-        buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:")
-        _stable_repr_to(buf, obj.__dict__, _visited)
-        buf.write(">")
-        _visited.discard(obj_id)
     else:
-        buf.write(repr(obj))
+        state = _object_state(obj)
+        if state is not None:
+            obj_id = id(obj)
+            if obj_id in _visited:
+                buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:...>")
+                return
+            _visited.add(obj_id)
+            buf.write(f"<{type(obj).__module__}.{type(obj).__qualname__}:")
+            _stable_repr_to(buf, state, _visited)
+            buf.write(">")
+            _visited.discard(obj_id)
+        else:
+            if type(obj).__repr__ is object.__repr__:
+                warnings.warn(
+                    f"Argument of type "
+                    f"{type(obj).__module__}.{type(obj).__qualname__} has no "
+                    f"__dict__/__slots__ and uses the default repr; it cannot be "
+                    f"hashed by value and will not cache reliably. Pass a "
+                    f"value-stable representation instead.",
+                    UnhashableArgWarning,
+                    stacklevel=2,
+                )
+            buf.write(repr(obj))
 
 
 def _stable_hash(
diff --git a/tests/test_hashing.py b/tests/test_hashing.py
index 23d2bd0..dd9b0a0 100644
--- a/tests/test_hashing.py
+++ b/tests/test_hashing.py
@@ -1,11 +1,18 @@
 from __future__ import annotations
 
 import warnings
+from dataclasses import dataclass
 from datetime import date
 from typing import Any
 
 from cashet import Client
-from cashet.hashing import ClosureWarning, _ast_canonical, hash_function
+from cashet.hashing import (
+    ClosureWarning,
+    UnhashableArgWarning,
+    _ast_canonical,
+    hash_args,
+    hash_function,
+)
 
 
 class TestHashingEdgeCases:
@@ -123,6 +130,71 @@ def f(x: int = base) -> int:
         assert ref2.load() == 2
 
 
+class TestObjectStateHashing:
+    def test_slotted_objects_with_equal_state_hash_equal(self) -> None:
+        class Point:
+            __slots__ = ("x", "y")
+
+            def __init__(self, x: int, y: int) -> None:
+                self.x = x
+                self.y = y
+
+        assert hash_args(Point(1, 2)) == hash_args(Point(1, 2))
+
+    def test_slotted_objects_with_different_state_hash_differ(self) -> None:
+        class Point:
+            __slots__ = ("x", "y")
+
+            def __init__(self, x: int, y: int) -> None:
+                self.x = x
+                self.y = y
+
+        assert hash_args(Point(1, 2)) != hash_args(Point(1, 3))
+
+    def test_dataclass_with_slots_hashes_by_value(self) -> None:
+        @dataclass(slots=True)
+        class Config:
+            name: str
+            limit: int
+
+        assert hash_args(Config("a", 1)) == hash_args(Config("a", 1))
+        assert hash_args(Config("a", 1)) != hash_args(Config("a", 2))
+
+    def test_mixed_dict_and_slots_state_both_hashed(self) -> None:
+        class Base:
+            __slots__ = ("__dict__", "a")
+
+            def __init__(self, a: int, b: int) -> None:
+                self.a = a
+                self.b = b
+
+        assert hash_args(Base(1, 2)) == hash_args(Base(1, 2))
+        assert hash_args(Base(1, 2)) != hash_args(Base(1, 9))
+
+    def test_opaque_object_warns(self) -> None:
+        class Opaque:
+            __slots__ = ()
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            hash_args(Opaque())
+        unhashable = [x for x in w if issubclass(x.category, UnhashableArgWarning)]
+        assert len(unhashable) >= 1
+
+    def test_custom_repr_object_does_not_warn(self) -> None:
+        class Stable:
+            __slots__ = ()
+
+            def __repr__(self) -> str:
+                return "Stable()"
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            hash_args(Stable())
+        unhashable = [x for x in w if issubclass(x.category, UnhashableArgWarning)]
+        assert len(unhashable) == 0
+
+
 class TestASTNormalizedHashing:
     def test_comment_stripped(self) -> None:
         src1 = "def foo(x):\n    # important\n    return x + 1"

From 932f15a22431ee0edb88e597258cff3d7c73d629 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:12:25 +0200
Subject: [PATCH 02/18] fix: track ResultRefs nested in custom-object
 attributes

_collect_input_refs descended dict/list/tuple/set/frozenset but not
custom-object state, while the hasher (_stable_repr_to) does. A ResultRef
held as an object attribute was hashed into the cache key but omitted from
commit lineage, so GC could evict an upstream blob the commit depends on.

_collect_input_refs now descends object state via the shared object_state
helper, matching the hasher. _object_state is promoted to public
object_state and memoized by type (lru_cache) since it now runs per-arg.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/dag.py     |  9 ++++++++-
 src/cashet/hashing.py | 10 ++++++----
 tests/test_core.py    | 21 +++++++++++++++++++++
 3 files changed, 35 insertions(+), 5 deletions(-)

diff --git a/src/cashet/dag.py b/src/cashet/dag.py
index 1151f11..6a9711c 100644
--- a/src/cashet/dag.py
+++ b/src/cashet/dag.py
@@ -6,7 +6,7 @@
 from datetime import UTC, datetime
 from typing import Any, Generic, TypeVar
 
-from cashet.hashing import Serializer
+from cashet.hashing import Serializer, object_state
 from cashet.models import Commit, ObjectRef, TaskDef, TaskStatus
 from cashet.protocols import AsyncStore
 
@@ -36,6 +36,13 @@ def _collect_input_refs(value: Any, refs: list[ObjectRef], visited: set[int]) ->
         for item in value:
             _collect_input_refs(item, refs, visited)
         visited.discard(value_id)
+    else:
+        state = object_state(value)
+        if state is not None:
+            visited.add(value_id)
+            for item in state.values():
+                _collect_input_refs(item, refs, visited)
+            visited.discard(value_id)
 
 
 def resolve_input_refs(args: tuple[Any, ...], kwargs: dict[str, Any]) -> list[ObjectRef]:
diff --git a/src/cashet/hashing.py b/src/cashet/hashing.py
index 63c2ebf..25a9680 100644
--- a/src/cashet/hashing.py
+++ b/src/cashet/hashing.py
@@ -12,6 +12,7 @@
 import types
 import warnings
 from datetime import timedelta
+from functools import lru_cache
 from typing import Any, Protocol, runtime_checkable
 
 from cashet.models import TaskDef
@@ -380,7 +381,8 @@ def hash_function(
     return h.hexdigest()
 
 
-def _slot_names(cls: type) -> list[str]:
+@lru_cache(maxsize=1024)
+def _slot_names(cls: type) -> tuple[str, ...]:
     names: list[str] = []
     for klass in cls.__mro__:
         slots = klass.__dict__.get("__slots__")
@@ -392,13 +394,13 @@ def _slot_names(cls: type) -> list[str]:
             if name in ("__dict__", "__weakref__"):
                 continue
             names.append(name)
-    return names
+    return tuple(names)
 
 
 _UNSET = object()
 
 
-def _object_state(obj: Any) -> dict[str, Any] | None:
+def object_state(obj: Any) -> dict[str, Any] | None:
     state: dict[str, Any] = {}
     instance_dict = getattr(obj, "__dict__", None)
     if isinstance(instance_dict, dict):
@@ -488,7 +490,7 @@ def _stable_repr_to(
     elif hasattr(obj, "__cashet_ref__"):
         buf.write(f"<ref:{obj.__cashet_ref__().hash}>")
     else:
-        state = _object_state(obj)
+        state = object_state(obj)
         if state is not None:
             obj_id = id(obj)
             if obj_id in _visited:
diff --git a/tests/test_core.py b/tests/test_core.py
index 840e8f8..39b1078 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -283,6 +283,27 @@ def total(values: list[int]) -> int:
         assert commit is not None
         assert [ref.hash for ref in commit.input_refs] == [data_ref.hash]
 
+    def test_ref_in_custom_object_attribute_recorded_as_input(self, client: Client) -> None:
+        class Box:
+            __slots__ = ("payload",)
+
+            def __init__(self, payload: object) -> None:
+                self.payload = payload
+
+        def gen_data() -> int:
+            return 42
+
+        def consume(box: object) -> int:
+            return 1
+
+        data_ref = client.submit(gen_data)
+        result_ref = client.submit(consume, Box(data_ref))
+        assert result_ref.load() == 1
+
+        commit = client.show(result_ref.commit_hash)
+        assert commit is not None
+        assert data_ref.hash in [ref.hash for ref in commit.input_refs]
+
     def test_chained_pipeline(self, client: Client) -> None:
         def step1() -> int:
             return 10

From e6f7a54fc6580af03b1f0579ce167bb791afb600 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:14:18 +0200
Subject: [PATCH 03/18] fix: hash referenced global dict/list/set consistently
 with scalars

A function referencing a module-level scalar invalidated its cache when the
scalar changed, but a referenced dict/list/set did not. That was a silent
stale-result bug and an undocumented inconsistency.

_should_hash_global_value now admits dict/list/set whose members are all
value-stable (recursively), so config-style globals invalidate like scalars.
Containers with unstable members (e.g. a custom object) remain unhashed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/hashing.py | 10 +++++++++-
 tests/test_hashing.py | 38 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/src/cashet/hashing.py b/src/cashet/hashing.py
index 25a9680..fe02bf6 100644
--- a/src/cashet/hashing.py
+++ b/src/cashet/hashing.py
@@ -297,11 +297,19 @@ def _should_hash_global_value(obj: Any, visited: set[int] | None = None) -> bool
         )
         visited.discard(obj_id)
         return result
-    if isinstance(obj, tuple | frozenset):
+    if isinstance(obj, tuple | frozenset | list | set):
         visited.add(obj_id)
         result = all(_should_hash_global_value(item, visited) for item in obj)
         visited.discard(obj_id)
         return result
+    if isinstance(obj, dict):
+        visited.add(obj_id)
+        result = all(
+            _should_hash_global_value(k, visited) and _should_hash_global_value(v, visited)
+            for k, v in obj.items()
+        )
+        visited.discard(obj_id)
+        return result
     return False
 
 
diff --git a/tests/test_hashing.py b/tests/test_hashing.py
index dd9b0a0..38a5805 100644
--- a/tests/test_hashing.py
+++ b/tests/test_hashing.py
@@ -493,6 +493,44 @@ def test_exec_function_invalidates_on_global_value_change(self, client: Client)
         assert ref1.load() == 20
         assert ref2.load() == 30
 
+    def test_exec_function_invalidates_on_dict_global_change(self, client: Client) -> None:
+        namespace: dict[str, Any] = {"CONFIG": {"factor": 2}}
+        exec("def f(x):\n    return x * CONFIG['factor']", namespace)
+        func = namespace["f"]
+        ref1 = client.submit(func, 10)
+
+        namespace["CONFIG"] = {"factor": 3}
+        ref2 = client.submit(func, 10)
+
+        assert ref1.hash != ref2.hash
+        assert ref1.load() == 20
+        assert ref2.load() == 30
+
+    def test_exec_function_invalidates_on_list_global_change(self, client: Client) -> None:
+        namespace: dict[str, Any] = {"WEIGHTS": [1, 2]}
+        exec("def f():\n    return sum(WEIGHTS)", namespace)
+        func = namespace["f"]
+        ref1 = client.submit(func)
+
+        namespace["WEIGHTS"] = [1, 2, 3]
+        ref2 = client.submit(func)
+
+        assert ref1.hash != ref2.hash
+        assert ref1.load() == 3
+        assert ref2.load() == 6
+
+    def test_global_container_with_unstable_member_not_hashed(self, client: Client) -> None:
+        namespace: dict[str, Any] = {"REGISTRY": {"handler": object()}}
+        exec("def f():\n    return len(REGISTRY)", namespace)
+        func = namespace["f"]
+        ref1 = client.submit(func)
+
+        namespace["REGISTRY"] = {"handler": object()}
+        ref2 = client.submit(func)
+
+        assert ref1.hash == ref2.hash
+        assert ref1.load() == 1
+
     def test_comprehension_invalidates_on_global_value_change(self, client: Client) -> None:
         namespace: dict[str, Any] = {"MULTIPLIER": 2}
         exec("def f(xs):\n    return [x * MULTIPLIER for x in xs]", namespace)

From 6379475df611ca7c9097ce4d457447faa704face Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:18:45 +0200
Subject: [PATCH 04/18] fix: tolerate a locked database when bumping
 last_accessed_at

find_by_fingerprint issued an UPDATE of last_accessed_at on every cache hit.
That write runs outside the write lock, so a concurrent writer holding the
lock past busy_timeout would raise OperationalError and turn a successful
cache lookup into a hard failure.

The bump only feeds LRU eviction ordering, so it is now best-effort: a locked
database is logged at debug and the cached commit is still returned.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/store.py | 13 +++++++++----
 tests/test_store.py | 30 ++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 4 deletions(-)

diff --git a/src/cashet/store.py b/src/cashet/store.py
index a80c60a..4901fac 100644
--- a/src/cashet/store.py
+++ b/src/cashet/store.py
@@ -341,10 +341,15 @@ def find_by_fingerprint(self, fingerprint: str) -> Commit | None:
         ).fetchone()
         if row is None:
             return None
-        conn.execute(
-            "UPDATE commits SET last_accessed_at = ? WHERE hash = ?",
-            (now_iso, row["hash"]),
-        )
+        try:
+            conn.execute(
+                "UPDATE commits SET last_accessed_at = ? WHERE hash = ?",
+                (now_iso, row["hash"]),
+            )
+        except sqlite3.OperationalError:
+            # Access-time bump only feeds LRU ordering; never fail a cache hit
+            # because a concurrent writer holds the lock past busy_timeout.
+            logger.debug("last_accessed_at bump skipped (db locked) hash=%s", row["hash"][:12])
         return self._row_to_commit(row)
 
     def find_running_by_fingerprint(self, fingerprint: str) -> Commit | None:
diff --git a/tests/test_store.py b/tests/test_store.py
index 13cee6b..3ec3afb 100644
--- a/tests/test_store.py
+++ b/tests/test_store.py
@@ -623,6 +623,36 @@ def test_inline_delete_removes_orphans(self, store_dir: Path) -> None:
         store.delete_commit(commit.hash)
         assert store.blob_exists(ref.hash) is False
 
+    def test_find_by_fingerprint_survives_locked_access_bump(self, tmp_path: Path) -> None:
+        from cashet.models import Commit, TaskDef, TaskStatus
+        from cashet.store import _SQLiteStoreCore
+
+        root = tmp_path / ".cashet"
+        core = _SQLiteStoreCore(root)
+        task_def = TaskDef(
+            func_hash="a", func_name="f", func_source="", args_hash="b", args_snapshot=b""
+        )
+        commit = Commit(
+            hash="c" * 64,
+            task_def=task_def,
+            output_ref=core.put_blob(b"v"),
+            status=TaskStatus.COMPLETED,
+        )
+        core.put_commit(commit)
+
+        conn = core._connect()
+        conn.execute("PRAGMA busy_timeout=100")
+        blocker = sqlite3.connect(str(root / "meta.db"), isolation_level=None)
+        blocker.execute("PRAGMA journal_mode=WAL")
+        blocker.execute("BEGIN IMMEDIATE")
+        try:
+            found = core.find_by_fingerprint(task_def.fingerprint)
+            assert found is not None
+            assert found.hash == commit.hash
+        finally:
+            blocker.execute("ROLLBACK")
+            blocker.close()
+
     def test_inline_stats(self, store_dir: Path) -> None:
         store = SQLiteStore(store_dir)
         store.put_blob(b"small")

From 48ca44f83764258e3309d8b180569049aa6be436 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:22:12 +0200
Subject: [PATCH 05/18] fix: do not fail eviction when post-eviction VACUUM is
 blocked

After deleting commits, evict ran VACUUM (or incremental_vacuum) unguarded.
A concurrent writer holding the lock makes VACUUM raise OperationalError,
which propagated out of evict even though the deletes had already committed.

Vacuum is extracted to _vacuum and wrapped in a best-effort guard: space
reclaim can be skipped, but a completed eviction must not surface as an error.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/store.py | 20 ++++++++++++++------
 tests/test_store.py | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 46 insertions(+), 6 deletions(-)

diff --git a/src/cashet/store.py b/src/cashet/store.py
index 4901fac..34f63c6 100644
--- a/src/cashet/store.py
+++ b/src/cashet/store.py
@@ -558,17 +558,25 @@ def evict(
                 deleted,
                 "size_limit" if max_size_bytes is not None else "ttl",
             )
-            c = self._connect()
-            mode = c.execute("PRAGMA auto_vacuum").fetchone()[0]
-            if mode == 2:
-                c.execute("PRAGMA incremental_vacuum")
-            else:
-                c.execute("VACUUM")
+            try:
+                self._vacuum()
+            except sqlite3.OperationalError:
+                # Reclaiming space is best-effort; a concurrent writer holding the
+                # lock must not undo an eviction whose deletes already committed.
+                logger.debug("vacuum skipped (db busy) after eviction")
         else:
             logger.debug("eviction found no candidates")
 
         return deleted
 
+    def _vacuum(self) -> None:
+        conn = self._connect()
+        mode = conn.execute("PRAGMA auto_vacuum").fetchone()[0]
+        if mode == 2:
+            conn.execute("PRAGMA incremental_vacuum")
+        else:
+            conn.execute("VACUUM")
+
     def _evict_to_size(self, current_bytes: int, max_size_bytes: int) -> int:
         pending_orphans: list[str] = []
         deleted = 0
diff --git a/tests/test_store.py b/tests/test_store.py
index 3ec3afb..576ef7a 100644
--- a/tests/test_store.py
+++ b/tests/test_store.py
@@ -653,6 +653,38 @@ def test_find_by_fingerprint_survives_locked_access_bump(self, tmp_path: Path) -
             blocker.execute("ROLLBACK")
             blocker.close()
 
+    def test_evict_survives_vacuum_failure(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from cashet.models import Commit, TaskDef, TaskStatus
+        from cashet.store import _SQLiteStoreCore
+
+        root = tmp_path / ".cashet"
+        core = _SQLiteStoreCore(root)
+        task_def = TaskDef(
+            func_hash="a", func_name="f", func_source="", args_hash="b", args_snapshot=b""
+        )
+        commit = Commit(
+            hash="c" * 64,
+            task_def=task_def,
+            output_ref=core.put_blob(b"v"),
+            status=TaskStatus.COMPLETED,
+        )
+        core.put_commit(commit)
+        core._connect().execute(
+            "UPDATE commits SET last_accessed_at = ? WHERE hash = ?",
+            ("2000-01-01T00:00:00+00:00", commit.hash),
+        )
+
+        def boom() -> None:
+            raise sqlite3.OperationalError("database is locked")
+
+        monkeypatch.setattr(core, "_vacuum", boom)
+
+        deleted = core.evict(datetime.now(UTC))
+        assert deleted == 1
+        assert core.get_commit(commit.hash) is None
+
     def test_inline_stats(self, store_dir: Path) -> None:
         store = SQLiteStore(store_dir)
         store.put_blob(b"small")

From 7b347c7d8ef542cecc75c5b05d859919ec3d7d2e Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:25:04 +0200
Subject: [PATCH 06/18] fix: prevent Redis tag index key collisions

_tag_key and _tag_value_key both built "cashet:tag:{...}", so a presence
index for tag key "a:b" collided with the value index for tag a="b", and a
':' inside any key or value could cross-match unrelated commits. delete_by_tags
then removed commits it should not have.

Presence and value indexes now use distinct prefixes (tagk:/tagv:) and the
value key length-prefixes the tag key, making the construction injective.

Note: this changes the Redis tag index key scheme. Tag indexes written by
older versions are not migrated; rewrite affected commits to rebuild them.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/redis_store.py |  6 ++++--
 tests/test_redis_store.py | 30 ++++++++++++++++++++++++++++++
 2 files changed, 34 insertions(+), 2 deletions(-)

diff --git a/src/cashet/redis_store.py b/src/cashet/redis_store.py
index 3bb0951..97002e9 100644
--- a/src/cashet/redis_store.py
+++ b/src/cashet/redis_store.py
@@ -63,11 +63,13 @@ def _status_key(status: str) -> str:
 
 
 def _tag_key(key: str) -> str:
-    return f"cashet:tag:{key}"
+    return f"cashet:tagk:{key}"
 
 
 def _tag_value_key(key: str, value: str) -> str:
-    return f"cashet:tag:{key}:{value}"
+    # Length-prefix the key so a ':' inside a tag key or value can never make
+    # two distinct (key, value) pairs collide onto the same set.
+    return f"cashet:tagv:{len(key)}:{key}:{value}"
 
 
 def _access_key() -> str:
diff --git a/tests/test_redis_store.py b/tests/test_redis_store.py
index c0f25c7..dd31ca2 100644
--- a/tests/test_redis_store.py
+++ b/tests/test_redis_store.py
@@ -580,6 +580,36 @@ def test_delete_by_tags_exact_match(self, redis_store: RedisStore) -> None:
         assert deleted == 1
         assert redis_store.get_commit("d" * 64) is None
 
+    def test_delete_by_tags_no_collision_with_colon_in_key(
+        self, redis_store: RedisStore
+    ) -> None:
+        # Tag key "a:b" (presence) must not collide with tag a="b" (value).
+        colon_key = Commit(
+            hash="a" * 64,
+            task_def=TaskDef(
+                func_hash="1" * 64, func_name="f", func_source="", args_hash="b" * 64,
+                args_snapshot=b"", tags={"a:b": "x"},
+            ),
+            tags={"a:b": "x"},
+            status=TaskStatus.COMPLETED,
+        )
+        value_pair = Commit(
+            hash="b" * 64,
+            task_def=TaskDef(
+                func_hash="2" * 64, func_name="f", func_source="", args_hash="b" * 64,
+                args_snapshot=b"", tags={"a": "b"},
+            ),
+            tags={"a": "b"},
+            status=TaskStatus.COMPLETED,
+        )
+        redis_store.put_commit(colon_key)
+        redis_store.put_commit(value_pair)
+
+        deleted = redis_store.delete_by_tags({"a:b": None})
+        assert deleted == 1
+        assert redis_store.get_commit("a" * 64) is None
+        assert redis_store.get_commit("b" * 64) is not None
+
     def test_delete_by_tags_bare_key(self, redis_store: RedisStore) -> None:
         for i, env in enumerate(["prod", "staging"]):
             task_def = TaskDef(

From 15dc57a3a28407be0009312b6e6c1d823c42f968 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:28:50 +0200
Subject: [PATCH 07/18] fix: make Redis blob delete idempotent against a
 missing ref counter

_DECR_DELETE_SCRIPT called DECR unconditionally. On a missing ref key DECR
returns -1, which is <= 0, so the script deleted the blob even though that
ref key was not tracking it. An inconsistent or already-cleaned counter could
therefore drop a blob another commit still references.

The script now returns early when the ref key does not exist, so the decrement
never goes negative and a blob is only removed when its own counter reaches
zero.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/redis_store.py |  3 +++
 tests/test_redis_store.py | 28 ++++++++++++++++++++++++++++
 2 files changed, 31 insertions(+)

diff --git a/src/cashet/redis_store.py b/src/cashet/redis_store.py
index 97002e9..d12e711 100644
--- a/src/cashet/redis_store.py
+++ b/src/cashet/redis_store.py
@@ -18,6 +18,9 @@
     local ref_key = KEYS[1]
     local blob_key = KEYS[2]
     local stats_key = KEYS[3]
+    if redis.call('EXISTS', ref_key) == 0 then
+        return 0
+    end
     local count = redis.call('DECR', ref_key)
     if count <= 0 then
         local existed = redis.call('EXISTS', blob_key)
diff --git a/tests/test_redis_store.py b/tests/test_redis_store.py
index dd31ca2..2e87c54 100644
--- a/tests/test_redis_store.py
+++ b/tests/test_redis_store.py
@@ -580,6 +580,34 @@ def test_delete_by_tags_exact_match(self, redis_store: RedisStore) -> None:
         assert deleted == 1
         assert redis_store.get_commit("d" * 64) is None
 
+    def test_delete_does_not_drop_blob_when_ref_counter_missing(
+        self, redis_store: RedisStore
+    ) -> None:
+        import redis as _redis
+
+        from cashet.redis_store import _blob_ref_key
+
+        shared = redis_store.put_blob(b"shared-payload")
+        for h, fh in (("a" * 64, "1" * 64), ("b" * 64, "2" * 64)):
+            redis_store.put_commit(
+                Commit(
+                    hash=h,
+                    task_def=TaskDef(
+                        func_hash=fh, func_name="f", func_source="",
+                        args_hash="b" * 64, args_snapshot=b"",
+                    ),
+                    output_ref=shared,
+                    status=TaskStatus.COMPLETED,
+                )
+            )
+        # Simulate a lost/inconsistent ref counter for the shared blob.
+        _redis.Redis().delete(_blob_ref_key(shared.hash))
+
+        redis_store.delete_commit("a" * 64)
+
+        # Blob must survive: commit "b" still references it.
+        assert redis_store.get_blob(shared) == b"shared-payload"
+
     def test_delete_by_tags_no_collision_with_colon_in_key(
         self, redis_store: RedisStore
     ) -> None:

From a1fc110ae3042b79933ed91ab3aa96d493a28e1f Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:33:49 +0200
Subject: [PATCH 08/18] fix: enforce server request size limit on the bytes
 actually received

The size guard only checked the Content-Length header, which is absent for
chunked requests. A client could stream an unbounded body with no
Content-Length and the handler would buffer all of it via request.json(),
defeating the only DoS control on the server.

The limit is now a pure-ASGI middleware that counts bytes as they arrive,
rejects with 413 once the cap is exceeded, and replays the buffered body to
the app so handlers still read it normally. The Content-Length fast path is
kept for early rejection.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/server.py | 82 ++++++++++++++++++++++++++++++--------------
 tests/test_server.py | 31 +++++++++++++++++
 2 files changed, 88 insertions(+), 25 deletions(-)

diff --git a/src/cashet/server.py b/src/cashet/server.py
index 1f35634..5327249 100644
--- a/src/cashet/server.py
+++ b/src/cashet/server.py
@@ -12,7 +12,6 @@
 
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
-from starlette.middleware.base import BaseHTTPMiddleware
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 from starlette.routing import Route
@@ -48,24 +47,10 @@ def _validate_remote_code_options(
 _DEFAULT_MAX_CONTENT_LENGTH = 500 * 1024 * 1024
 
 
-def _limit_request_size(
-    request: Request,
-) -> JSONResponse | None:
-    max_size = getattr(request.app.state, "max_content_length", _DEFAULT_MAX_CONTENT_LENGTH)
-    content_length = request.headers.get("content-length")
-    if content_length is not None:
-        try:
-            size = int(content_length)
-        except ValueError:
-            return _CustomJSONResponse(
-                {"error": "invalid content-length"}, status_code=400
-            )
-        if size > max_size:
-            return _CustomJSONResponse(
-                {"error": f"request body exceeds {max_size} bytes"},
-                status_code=413,
-            )
-    return None
+def _too_large_response(max_size: int) -> _CustomJSONResponse:
+    return _CustomJSONResponse(
+        {"error": f"request body exceeds {max_size} bytes"}, status_code=413
+    )
 
 
 def _reconstruct_func(data: dict[str, Any]) -> Callable[..., Any] | None:
@@ -373,12 +358,59 @@ def create_async_app(
     return app
 
 
-class _SizeLimitMiddleware(BaseHTTPMiddleware):
-    async def dispatch(self, request: Request, call_next: Any) -> JSONResponse:
-        error = _limit_request_size(request)
-        if error is not None:
-            return error
-        return await call_next(request)
+class _SizeLimitMiddleware:
+    # Pure-ASGI so the body cap is enforced on bytes actually received, not on a
+    # client-supplied Content-Length that is absent for chunked requests.
+    def __init__(self, app: Any) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        starlette_app = scope.get("app")
+        max_size = (
+            getattr(starlette_app.state, "max_content_length", _DEFAULT_MAX_CONTENT_LENGTH)
+            if starlette_app is not None
+            else _DEFAULT_MAX_CONTENT_LENGTH
+        )
+        for name, value in scope["headers"]:
+            if name == b"content-length":
+                try:
+                    declared = int(value)
+                except ValueError:
+                    await _CustomJSONResponse(
+                        {"error": "invalid content-length"}, status_code=400
+                    )(scope, receive, send)
+                    return
+                if declared > max_size:
+                    await _too_large_response(max_size)(scope, receive, send)
+                    return
+                break
+
+        body = bytearray()
+        while True:
+            message = await receive()
+            if message["type"] != "http.request":
+                break
+            body.extend(message.get("body", b""))
+            if len(body) > max_size:
+                await _too_large_response(max_size)(scope, receive, send)
+                return
+            if not message.get("more_body", False):
+                break
+
+        buffered = bytes(body)
+        replayed = False
+
+        async def replay_receive() -> Any:
+            nonlocal replayed
+            if not replayed:
+                replayed = True
+                return {"type": "http.request", "body": buffered, "more_body": False}
+            return await receive()
+
+        await self.app(scope, replay_receive, send)
 
 
 # Sync handlers run sync Client operations in threads so they don't block the event loop
diff --git a/tests/test_server.py b/tests/test_server.py
index e527bff..bef715b 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import base64
+from collections.abc import Iterator
 from pathlib import Path
 
 import httpx
@@ -44,6 +45,36 @@ def server_client(tmp_path: Path) -> TestClient:
     return TestClient(app)
 
 
+class TestServerSizeLimit:
+    def _small_app(self, tmp_path: Path) -> TestClient:
+        client = Client(store_dir=tmp_path / ".cashet")
+        app = create_app(client, tasks={"add": _add}, max_content_length=200)
+        return TestClient(app)
+
+    def test_content_length_over_limit_rejected(self, tmp_path: Path) -> None:
+        tc = self._small_app(tmp_path)
+        payload = {"task": "add", "args": [1, 2], "pad": "x" * 500}
+        response = tc.post("/submit", json=payload)
+        assert response.status_code == 413
+        assert "exceeds" in response.json()["error"]
+
+    def test_chunked_request_without_content_length_rejected(self, tmp_path: Path) -> None:
+        tc = self._small_app(tmp_path)
+
+        def body_chunks() -> Iterator[bytes]:
+            for _ in range(10):
+                yield b"x" * 100  # 1000 bytes total, no Content-Length
+
+        response = tc.post("/submit", content=body_chunks())
+        assert response.status_code == 413
+        assert "exceeds" in response.json()["error"]
+
+    def test_request_within_limit_ok(self, tmp_path: Path) -> None:
+        tc = self._small_app(tmp_path)
+        response = tc.post("/submit", json={"task": "add", "args": [3, 4], "kwargs": {}})
+        assert response.status_code == 200
+
+
 class TestServerSubmit:
     def test_submit_registered_task_with_json_args(self, server_client: TestClient) -> None:
         payload = {"task": "add", "args": [3, 4], "kwargs": {}}

From e552656e930f45944fb3e70301174ba68baefe7f Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:39:10 +0200
Subject: [PATCH 09/18] fix: validate input and add error barriers to all
 server handlers

Several handlers parsed input without guards: ?limit=abc raised ValueError and
/gc with a bad type or non-JSON body raised, all surfacing as uncaught 500s
(the async commit/log/stats/gc handlers had no exception barrier at all, so
internals could leak through Starlette's default 500).

Adds a shared _safe_handler barrier applied to every route that maps bad input
to 400 and any unexpected error to a generic 500, a _query_int helper for query
params, and explicit gc parameter validation. /gc now also accepts an empty
body and falls back to defaults.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/server.py | 107 +++++++++++++++++++++++++++++++++++--------
 tests/test_server.py |  56 ++++++++++++++++++++++
 2 files changed, 143 insertions(+), 20 deletions(-)

diff --git a/src/cashet/server.py b/src/cashet/server.py
index 5327249..a86e4ce 100644
--- a/src/cashet/server.py
+++ b/src/cashet/server.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import base64
+import functools
 import hmac
 import json
 import logging
@@ -53,6 +54,60 @@ def _too_large_response(max_size: int) -> _CustomJSONResponse:
     )
 
 
+class _BadRequestError(Exception):
+    pass
+
+
+def _query_int(request: Request, name: str, default: int) -> int:
+    raw = request.query_params.get(name)
+    if raw is None:
+        return default
+    try:
+        return int(raw)
+    except ValueError as exc:
+        raise _BadRequestError(f"{name} must be an integer") from exc
+
+
+async def _json_body(request: Request) -> dict[str, Any]:
+    body = await request.body()
+    if not body:
+        return {}
+    data = json.loads(body)
+    if not isinstance(data, dict):
+        raise _BadRequestError("request body must be a JSON object")
+    return data
+
+
+def _gc_params(data: dict[str, Any]) -> tuple[float, int | None]:
+    older_than_days = data.get("older_than_days", 30)
+    if isinstance(older_than_days, bool) or not isinstance(older_than_days, int | float):
+        raise _BadRequestError("older_than_days must be a number")
+    max_size = data.get("max_size")
+    if max_size is not None and (isinstance(max_size, bool) or not isinstance(max_size, int)):
+        raise _BadRequestError("max_size must be an integer or null")
+    return older_than_days, max_size
+
+
+def _safe_handler(handler: Any) -> Any:
+    @functools.wraps(handler)
+    async def wrapper(request: Request) -> JSONResponse:
+        try:
+            return await handler(request)
+        except _BadRequestError as exc:
+            return _CustomJSONResponse({"error": str(exc)}, status_code=400)
+        except json.JSONDecodeError:
+            return _CustomJSONResponse({"error": "invalid JSON body"}, status_code=400)
+        except Exception:
+            logger.exception(
+                "request failed method=%s path=%s status=500",
+                request.method,
+                request.url.path,
+            )
+            return _CustomJSONResponse({"error": "Internal server error"}, status_code=500)
+
+    return wrapper
+
+
 def _reconstruct_func(data: dict[str, Any]) -> Callable[..., Any] | None:
     func_b64 = data.get("func_b64")
     func_source = data.get("func_source")
@@ -280,7 +335,7 @@ async def _async_commit(request: Request) -> JSONResponse:
 async def _async_log(request: Request) -> JSONResponse:
     client: AsyncClient = request.app.state.client
     func_name = request.query_params.get("func")
-    limit = int(request.query_params.get("limit", "50"))
+    limit = _query_int(request, "limit", 50)
     status = request.query_params.get("status")
     start = time.perf_counter()
     commits = await client.log(func_name=func_name, limit=limit, status=status)
@@ -308,9 +363,7 @@ async def _async_gc(request: Request) -> JSONResponse:
     from datetime import timedelta
 
     client: AsyncClient = request.app.state.client
-    data = await request.json()
-    older_than_days = data.get("older_than_days", 30)
-    max_size = data.get("max_size")
+    older_than_days, max_size = _gc_params(await _json_body(request))
     start = time.perf_counter()
     deleted = await client.gc(timedelta(days=older_than_days), max_size_bytes=max_size)
     duration = int((time.perf_counter() - start) * 1000)
@@ -331,20 +384,28 @@ def create_async_app(
 ) -> Starlette:
     _validate_remote_code_options(allow_remote_code, require_token)
     routes = [
-        Route("/submit", _require_token(_async_submit, require_token), methods=["POST"]),
+        Route(
+            "/submit",
+            _require_token(_safe_handler(_async_submit), require_token),
+            methods=["POST"],
+        ),
         Route(
             "/result/{commit_hash}",
-            _require_token(_async_result, require_token),
+            _require_token(_safe_handler(_async_result), require_token),
             methods=["GET"],
         ),
         Route(
             "/commit/{commit_hash}",
-            _require_token(_async_commit, require_token),
+            _require_token(_safe_handler(_async_commit), require_token),
+            methods=["GET"],
+        ),
+        Route("/log", _require_token(_safe_handler(_async_log), require_token), methods=["GET"]),
+        Route(
+            "/stats",
+            _require_token(_safe_handler(_async_stats), require_token),
             methods=["GET"],
         ),
-        Route("/log", _require_token(_async_log, require_token), methods=["GET"]),
-        Route("/stats", _require_token(_async_stats, require_token), methods=["GET"]),
-        Route("/gc", _require_token(_async_gc, require_token), methods=["POST"]),
+        Route("/gc", _require_token(_safe_handler(_async_gc), require_token), methods=["POST"]),
     ]
 
     app = Starlette(
@@ -534,7 +595,7 @@ def _run() -> JSONResponse:
 async def _log(request: Request) -> JSONResponse:
     client: Client = request.app.state.client
     func_name = request.query_params.get("func")
-    limit = int(request.query_params.get("limit", "50"))
+    limit = _query_int(request, "limit", 50)
     status = request.query_params.get("status")
     start = time.perf_counter()
 
@@ -569,9 +630,7 @@ async def _gc(request: Request) -> JSONResponse:
     from datetime import timedelta
 
     client: Client = request.app.state.client
-    data = await request.json()
-    older_than_days = data.get("older_than_days", 30)
-    max_size = data.get("max_size")
+    older_than_days, max_size = _gc_params(await _json_body(request))
     start = time.perf_counter()
 
     def _run() -> JSONResponse:
@@ -597,20 +656,28 @@ def create_app(
 ) -> Starlette:
     _validate_remote_code_options(allow_remote_code, require_token)
     routes = [
-        Route("/submit", _require_token(_submit, require_token), methods=["POST"]),
+        Route(
+            "/submit",
+            _require_token(_safe_handler(_submit), require_token),
+            methods=["POST"],
+        ),
         Route(
             "/result/{commit_hash}",
-            _require_token(_result, require_token),
+            _require_token(_safe_handler(_result), require_token),
             methods=["GET"],
         ),
         Route(
             "/commit/{commit_hash}",
-            _require_token(_commit, require_token),
+            _require_token(_safe_handler(_commit), require_token),
+            methods=["GET"],
+        ),
+        Route("/log", _require_token(_safe_handler(_log), require_token), methods=["GET"]),
+        Route(
+            "/stats",
+            _require_token(_safe_handler(_stats), require_token),
             methods=["GET"],
         ),
-        Route("/log", _require_token(_log, require_token), methods=["GET"]),
-        Route("/stats", _require_token(_stats, require_token), methods=["GET"]),
-        Route("/gc", _require_token(_gc, require_token), methods=["POST"]),
+        Route("/gc", _require_token(_safe_handler(_gc), require_token), methods=["POST"]),
     ]
     app = Starlette(
         routes=routes,
diff --git a/tests/test_server.py b/tests/test_server.py
index bef715b..53230c1 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -75,6 +75,62 @@ def test_request_within_limit_ok(self, tmp_path: Path) -> None:
         assert response.status_code == 200
 
 
+class TestServerValidation:
+    def test_log_bad_limit_returns_400(self, server_client: TestClient) -> None:
+        response = server_client.get("/log?limit=abc")
+        assert response.status_code == 400
+        assert "limit" in response.json()["error"]
+
+    def test_gc_bad_older_than_days_returns_400(self, server_client: TestClient) -> None:
+        response = server_client.post("/gc", json={"older_than_days": "soon"})
+        assert response.status_code == 400
+
+    def test_gc_invalid_json_returns_400(self, server_client: TestClient) -> None:
+        response = server_client.post(
+            "/gc", content=b"{not json", headers={"content-type": "application/json"}
+        )
+        assert response.status_code == 400
+
+    def test_gc_empty_body_uses_defaults(self, server_client: TestClient) -> None:
+        response = server_client.post("/gc")
+        assert response.status_code == 200
+        assert "deleted" in response.json()
+
+    def test_internal_error_returns_generic_500(
+        self, server_client: TestClient, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        client: Client = server_client.app.state.client  # type: ignore[attr-defined]
+
+        def boom() -> dict[str, int]:
+            raise RuntimeError("secret internal detail")
+
+        monkeypatch.setattr(client, "stats", boom)
+        response = server_client.get("/stats")
+        assert response.status_code == 500
+        assert response.json() == {"error": "Internal server error"}
+
+    async def test_async_log_bad_limit_returns_400(self, tmp_path: Path) -> None:
+        client = AsyncClient(store_dir=tmp_path / ".cashet")
+        app = create_async_app(client, tasks={"add": _add})
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            response = await ac.get("/log?limit=abc")
+        assert response.status_code == 400
+        await client.close()
+
+    async def test_async_gc_empty_body_uses_defaults(self, tmp_path: Path) -> None:
+        client = AsyncClient(store_dir=tmp_path / ".cashet")
+        app = create_async_app(client, tasks={"add": _add})
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as ac:
+            response = await ac.post("/gc")
+        assert response.status_code == 200
+        assert "deleted" in response.json()
+        await client.close()
+
+
 class TestServerSubmit:
     def test_submit_registered_task_with_json_args(self, server_client: TestClient) -> None:
         payload = {"task": "add", "args": [3, 4], "kwargs": {}}

From db5ff24e57f42640b1f9ecdadbf24d2b12e8d65d Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:44:18 +0200
Subject: [PATCH 10/18] fix: report commits skipped during a lossy import

import_store silently dropped any commit whose blobs were absent from the
archive and returned only the imported count, so a truncated archive looked
like a clean import. It now returns an ImportResult with imported and skipped
counts; import_archive on both clients returns it and the CLI prints the
skipped count.

This changes the import_archive return type from int to ImportResult (a
NamedTuple), so callers read result.imported instead of the bare count.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/_export.py      | 16 +++++++++---
 src/cashet/async_client.py |  4 +--
 src/cashet/cli.py          |  8 ++++--
 src/cashet/client.py       |  3 ++-
 tests/test_export.py       | 52 ++++++++++++++++++++++++++++++--------
 5 files changed, 65 insertions(+), 18 deletions(-)

diff --git a/src/cashet/_export.py b/src/cashet/_export.py
index 0d5a77c..e439f3a 100644
--- a/src/cashet/_export.py
+++ b/src/cashet/_export.py
@@ -8,7 +8,7 @@
 from datetime import datetime, timedelta
 from io import BytesIO
 from pathlib import Path
-from typing import Any
+from typing import Any, NamedTuple
 
 from cashet.models import Commit, ObjectRef, StorageTier, TaskDef, TaskStatus
 from cashet.protocols import AsyncStore
@@ -164,8 +164,14 @@ def _is_safe_tar_member(name: str) -> bool:
     return ".." not in name.split("/")
 
 
-async def import_store(store: AsyncStore, tar_path: Path) -> int:
+class ImportResult(NamedTuple):
+    imported: int
+    skipped: int
+
+
+async def import_store(store: AsyncStore, tar_path: Path) -> ImportResult:
     imported = 0
+    skipped = 0
     with tarfile.open(tar_path, "r:gz") as tar:
         members: dict[str, tarfile.TarInfo] = {}
         for m in tar.getmembers():
@@ -257,6 +263,10 @@ async def import_store(store: AsyncStore, tar_path: Path) -> int:
                         missing_blobs.add(h)
                         has_missing = True
             if has_missing:
+                skipped += 1
+                logger.warning(
+                    "skipping commit with missing blob(s) hash=%s", commit.hash[:12]
+                )
                 continue
             commit.input_refs = [
                 blob_refs.get(r.hash, r) for r in commit.input_refs
@@ -267,4 +277,4 @@ async def import_store(store: AsyncStore, tar_path: Path) -> int:
                     commit.output_ref = blob_refs[out_h]
             await store.put_commit(commit)
             imported += 1
-    return imported
+    return ImportResult(imported=imported, skipped=skipped)
diff --git a/src/cashet/async_client.py b/src/cashet/async_client.py
index c1c9499..deb8696 100644
--- a/src/cashet/async_client.py
+++ b/src/cashet/async_client.py
@@ -21,7 +21,7 @@
     resolve_task_config,
     set_task_metadata,
 )
-from cashet._export import export_store, import_store
+from cashet._export import ImportResult, export_store, import_store
 from cashet.async_executor import AsyncLocalExecutor
 from cashet.dag import AsyncResultRef
 from cashet.hashing import PickleSerializer, Serializer, build_task_def, warn_default_pickle
@@ -251,7 +251,7 @@ async def clear(self) -> int:
     async def export(self, path: str | Path) -> None:
         await export_store(self.store, Path(path))
 
-    async def import_archive(self, path: str | Path) -> int:
+    async def import_archive(self, path: str | Path) -> ImportResult:
         return await import_store(self.store, Path(path))
 
     async def close(self) -> None:
diff --git a/src/cashet/cli.py b/src/cashet/cli.py
index 1c2c13c..4d9f654 100644
--- a/src/cashet/cli.py
+++ b/src/cashet/cli.py
@@ -357,11 +357,15 @@ def import_archive(path: str) -> None:
 
     client = _client()
     try:
-        count = client.import_archive(path)
+        result = client.import_archive(path)
     except (OSError, tarfile.TarError, ValueError) as e:
         console.print(f"[red]Import failed: {e}[/red]")
         raise SystemExit(1) from None
-    console.print(f"[green]Imported {count} commit(s) from {path}[/green]")
+    console.print(f"[green]Imported {result.imported} commit(s) from {path}[/green]")
+    if result.skipped:
+        console.print(
+            f"[yellow]Skipped {result.skipped} commit(s) with missing blobs[/yellow]"
+        )
 
 
 if __name__ == "__main__":
diff --git a/src/cashet/client.py b/src/cashet/client.py
index 879255d..6c83cda 100644
--- a/src/cashet/client.py
+++ b/src/cashet/client.py
@@ -19,6 +19,7 @@
     resolve_store_dir,
     set_task_metadata,
 )
+from cashet._export import ImportResult
 from cashet._runner import BlockingAsyncRunner
 from cashet.adapters import SyncStoreAdapter
 from cashet.async_client import AsyncClient
@@ -314,7 +315,7 @@ def serve(
     def export(self, path: str | Path) -> None:
         self._runner.call(self._async_client.export(path))
 
-    def import_archive(self, path: str | Path) -> int:
+    def import_archive(self, path: str | Path) -> ImportResult:
         return self._runner.call(self._async_client.import_archive(path))
 
     def close(self) -> None:
diff --git a/tests/test_export.py b/tests/test_export.py
index a6036d6..cac79df 100644
--- a/tests/test_export.py
+++ b/tests/test_export.py
@@ -40,6 +40,22 @@ def corrupt_first_blob(src: Path, dst: Path) -> None:
             patched.addfile(info, BytesIO(data))
 
 
+def drop_first_blob_and_manifest(src: Path, dst: Path) -> None:
+    with tarfile.open(src, "r:gz") as original, tarfile.open(dst, "w:gz") as patched:
+        dropped = False
+        for member in original.getmembers():
+            if member.name == "cashet-export/manifest.json":
+                continue
+            if member.name.startswith("cashet-export/blobs/") and not dropped:
+                dropped = True
+                continue
+            extracted = original.extractfile(member) if member.isfile() else None
+            data = extracted.read() if extracted is not None else b""
+            info = tarfile.TarInfo(member.name)
+            info.size = len(data)
+            patched.addfile(info, BytesIO(data))
+
+
 class TestSyncExport:
     def test_export_import_roundtrip(self, client: Client, tmp_path: Path) -> None:
         ref1 = client.submit(add, 1, 2)
@@ -52,8 +68,8 @@ def test_export_import_roundtrip(self, client: Client, tmp_path: Path) -> None:
         assert archive.exists()
 
         client2 = Client(store_dir=tmp_path / ".cashet2")
-        count = client2.import_archive(archive)
-        assert count == 2
+        result = client2.import_archive(archive)
+        assert result.imported == 2
 
         imported1 = client2.show(ref1.commit_hash)
         assert imported1 is not None
@@ -73,8 +89,8 @@ def test_import_skips_existing(self, client: Client, tmp_path: Path) -> None:
         archive = tmp_path / "export.tar.gz"
         client.export(archive)
 
-        count = client.import_archive(archive)
-        assert count == 0
+        result = client.import_archive(archive)
+        assert result.imported == 0
 
     def test_export_empty_store(self, client: Client, tmp_path: Path) -> None:
         archive = tmp_path / "empty.tar.gz"
@@ -82,8 +98,8 @@ def test_export_empty_store(self, client: Client, tmp_path: Path) -> None:
         assert archive.exists()
 
         client2 = Client(store_dir=tmp_path / ".cashet2")
-        count = client2.import_archive(archive)
-        assert count == 0
+        result = client2.import_archive(archive)
+        assert result.imported == 0
 
     def test_import_rejects_corrupt_blob_payload(self, client: Client, tmp_path: Path) -> None:
         ref = client.submit(add, 1, 2)
@@ -99,6 +115,22 @@ def test_import_rejects_corrupt_blob_payload(self, client: Client, tmp_path: Pat
             client2.import_archive(corrupt_archive)
         assert client2.show(ref.commit_hash) is None
 
+    def test_import_reports_skipped_commits_with_missing_blobs(
+        self, client: Client, tmp_path: Path
+    ) -> None:
+        client.submit(add, 1, 2).load()
+        client.submit(greet, "world").load()
+
+        archive = tmp_path / "export.tar.gz"
+        lossy = tmp_path / "lossy.tar.gz"
+        client.export(archive)
+        drop_first_blob_and_manifest(archive, lossy)
+
+        client2 = Client(store_dir=tmp_path / ".cashet2")
+        result = client2.import_archive(lossy)
+        assert result.imported == 1
+        assert result.skipped == 1
+
 
 class TestAsyncExport:
     async def test_export_import_roundtrip(
@@ -114,8 +146,8 @@ async def test_export_import_roundtrip(
         assert archive.exists()
 
         client2 = AsyncClient(store_dir=tmp_path / ".cashet2")
-        count = await client2.import_archive(archive)
-        assert count == 2
+        result = await client2.import_archive(archive)
+        assert result.imported == 2
 
         imported1 = await client2.show(ref1.commit_hash)
         assert imported1 is not None
@@ -136,5 +168,5 @@ async def test_import_skips_existing(self, async_client: AsyncClient, tmp_path:
         archive = tmp_path / "export.tar.gz"
         await async_client.export(archive)
 
-        count = await async_client.import_archive(archive)
-        assert count == 0
+        result = await async_client.import_archive(archive)
+        assert result.imported == 0

From 1a6bd6731ebc0251755bad0bb543dc10ee2f986f Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:46:26 +0200
Subject: [PATCH 11/18] fix: exit non-zero from CLI show/get when a commit is
 missing

show and get printed a not-found message but exited 0, while rm exited 1 for
the same condition, so scripts could not detect a miss via the exit code.
show and get now exit 1 on not-found, matching rm.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/cli.py | 4 ++--
 tests/test_cli.py | 4 ++--
 2 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/src/cashet/cli.py b/src/cashet/cli.py
index 4d9f654..b19f3d2 100644
--- a/src/cashet/cli.py
+++ b/src/cashet/cli.py
@@ -119,7 +119,7 @@ def show_cmd(hash: str) -> None:
         raise SystemExit(1) from None
     if commit is None:
         console.print(f"[red]Commit {hash} not found.[/red]")
-        return
+        raise SystemExit(1)
     out_hash = commit.output_ref.hash[:12] if commit.output_ref else "none"
     out_size = commit.output_ref.size if commit.output_ref else 0
     parent = commit.parent_hash[:12] if commit.parent_hash else "none"
@@ -164,7 +164,7 @@ def get_cmd(hash: str, output: str | None) -> None:
         raise SystemExit(1) from None
     if commit is None or commit.output_ref is None:
         console.print(f"[red]No result for {hash}[/red]")
-        return
+        raise SystemExit(1)
     ref = client.store.get_blob(commit.output_ref)
     if output:
         pathlib.Path(output).write_bytes(ref)
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 9c6ff88..c60b329 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -104,7 +104,7 @@ def greet(name: str) -> str:
 
     def test_show_missing(self, cli_runner: CliRunner, store_dir: Path) -> None:
         result = _invoke(cli_runner, ["show", "deadbeef"], store_dir)
-        assert result.exit_code == 0
+        assert result.exit_code == 1
         assert "not found" in result.output
 
     def test_show_ambiguous_hash(self, cli_runner: CliRunner, store_dir: Path) -> None:
@@ -152,7 +152,7 @@ def val() -> str:
 
     def test_get_missing(self, cli_runner: CliRunner, store_dir: Path) -> None:
         result = _invoke(cli_runner, ["get", "deadbeef"], store_dir)
-        assert result.exit_code == 0
+        assert result.exit_code == 1
         assert "No result" in result.output
 
     def test_get_ambiguous_hash(self, cli_runner: CliRunner, store_dir: Path) -> None:

From 7efffa94a029c93628d578ba622e923c145a4b2a Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:48:56 +0200
Subject: [PATCH 12/18] fix: canonicalize function source with ast.unparse for
 cross-version stability

_ast_canonical hashed ast.dump output, whose field set changes across Python
versions (for example type_params added in 3.12), so the same function hashed
differently on different interpreters and missed the cache in distributed
deployments mixing versions.

It now hashes ast.unparse output, which keeps the comment, whitespace, and
docstring insensitivity but is source text and far more stable across versions.
Docstring stripping inserts a pass when it would leave an empty body so
unparse stays valid.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/hashing.py |  7 ++++++-
 tests/test_hashing.py | 13 +++++++++++++
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/src/cashet/hashing.py b/src/cashet/hashing.py
index fe02bf6..ffb893b 100644
--- a/src/cashet/hashing.py
+++ b/src/cashet/hashing.py
@@ -208,10 +208,13 @@ def hash_source(source: str) -> str:
 
 
 def _ast_canonical(source: str) -> str:
+    # ast.unparse normalizes whitespace and comments like ast.dump but, being
+    # source text rather than the internal AST repr, stays stable across Python
+    # versions whose ast.dump field set differs (e.g. type_params in 3.12).
     try:
         tree = ast.parse(source)
         _strip_docstrings(tree)
-        return ast.dump(tree)
+        return ast.unparse(tree)
     except SyntaxError:
         return source
 
@@ -230,6 +233,8 @@ def _strip_docstrings(node: ast.AST) -> None:
             and isinstance(child.body[0].value.value, str)
         ):
             child.body = child.body[1:]
+            if not child.body:
+                child.body = [ast.Pass()]
 
 
 def _is_stdlib_or_site_path(path: str) -> bool:
diff --git a/tests/test_hashing.py b/tests/test_hashing.py
index 38a5805..0ad23e0 100644
--- a/tests/test_hashing.py
+++ b/tests/test_hashing.py
@@ -216,6 +216,19 @@ def test_docstring_stripped(self) -> None:
         src2 = 'def foo(x):\n    """updated docs"""\n    return x'
         assert _ast_canonical(src1) == _ast_canonical(src2)
 
+    def test_only_docstring_function_does_not_crash(self) -> None:
+        src = 'def f():\n    """only docs"""'
+        out = _ast_canonical(src)
+        assert "only docs" not in out
+        assert _ast_canonical(src) == out
+
+    def test_canonical_form_is_parseable_source(self) -> None:
+        import ast
+
+        out = _ast_canonical("def f(x):\n    # note\n    return x+1")
+        # ast.unparse yields source text (round-trippable), unlike ast.dump.
+        ast.parse(out)
+
     def test_different_functions_different_hash(self) -> None:
         def add(x: int, y: int) -> int:
             return x + y

From 4c3a8e690d7e94cf0adf8a156b8f809ea6eccae6 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:51:54 +0200
Subject: [PATCH 13/18] perf: O(1) running-claim lookup in Redis via a
 per-fingerprint index

find_running_by_fingerprint scanned every commit ever recorded for a
fingerprint and called get_commit on each, so submits on a hot fingerprint
(for example accumulated cache=False salted runs) cost O(history) per call
inside the claim lock.

A per-fingerprint running set is now maintained: a hash is added on a RUNNING
put and removed on any other status, delete, or eviction. The lookup reads
that set, which holds at most the in-flight claims. Each index update only
touches the commit's own hash, so the set stays correct.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/redis_store.py | 11 ++++++++++-
 tests/test_redis_store.py | 15 +++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/src/cashet/redis_store.py b/src/cashet/redis_store.py
index d12e711..334230e 100644
--- a/src/cashet/redis_store.py
+++ b/src/cashet/redis_store.py
@@ -57,6 +57,10 @@ def _fp_key(fingerprint: str) -> str:
     return f"cashet:index:fingerprint:{fingerprint}"
 
 
+def _running_key(fingerprint: str) -> str:
+    return f"cashet:index:running:{fingerprint}"
+
+
 def _func_key(func_name: str) -> str:
     return f"cashet:index:func:{func_name}"
 
@@ -255,6 +259,10 @@ def _index_commit_commands(pipe: Any, commit: Commit) -> None:
     for status in TaskStatus:
         pipe.srem(_status_key(status.value), commit.hash)
     pipe.sadd(_status_key(commit.status.value), commit.hash)
+    if commit.status == TaskStatus.RUNNING:
+        pipe.sadd(_running_key(commit.fingerprint), commit.hash)
+    else:
+        pipe.srem(_running_key(commit.fingerprint), commit.hash)
     for key, val in commit.tags.items():
         pipe.sadd(_tag_key(key), commit.hash)
         pipe.sadd(_tag_value_key(key, val), commit.hash)
@@ -264,6 +272,7 @@ def _remove_commit_index_commands(pipe: Any, commit: Commit, resolved_hash: str)
     pipe.delete(_commit_key(resolved_hash))
     pipe.zrem("cashet:index:all", resolved_hash)
     pipe.zrem(_fp_key(commit.fingerprint), resolved_hash)
+    pipe.srem(_running_key(commit.fingerprint), resolved_hash)
     pipe.zrem(_func_key(commit.task_def.func_name), resolved_hash)
     pipe.zrem(_access_key(), resolved_hash)
     for status in TaskStatus:
@@ -402,7 +411,7 @@ async def find_by_fingerprint(self, fingerprint: str) -> Commit | None:
         return None
 
     async def find_running_by_fingerprint(self, fingerprint: str) -> Commit | None:
-        hashes = await self._redis.zrevrange(_fp_key(fingerprint), 0, -1)
+        hashes = await self._redis.smembers(_running_key(fingerprint))
         for h in hashes:
             h_str = h.decode() if isinstance(h, bytes) else h
             commit = await self.get_commit(h_str)
diff --git a/tests/test_redis_store.py b/tests/test_redis_store.py
index 2e87c54..f35683a 100644
--- a/tests/test_redis_store.py
+++ b/tests/test_redis_store.py
@@ -580,6 +580,21 @@ def test_delete_by_tags_exact_match(self, redis_store: RedisStore) -> None:
         assert deleted == 1
         assert redis_store.get_commit("d" * 64) is None
 
+    def test_find_running_tracks_running_index(self, redis_store: RedisStore) -> None:
+        task_def = TaskDef(
+            func_hash="a" * 64, func_name="f", func_source="",
+            args_hash="b" * 64, args_snapshot=b"",
+        )
+        running = Commit(hash="a" * 64, task_def=task_def, status=TaskStatus.RUNNING)
+        redis_store.put_commit(running)
+        found = redis_store.find_running_by_fingerprint(task_def.fingerprint)
+        assert found is not None and found.hash == "a" * 64
+
+        # transition out of RUNNING clears the running claim
+        running.status = TaskStatus.COMPLETED
+        redis_store.put_commit(running)
+        assert redis_store.find_running_by_fingerprint(task_def.fingerprint) is None
+
     def test_delete_does_not_drop_blob_when_ref_counter_missing(
         self, redis_store: RedisStore
     ) -> None:

From 8c8f99743ed27509db0c3a2af157a146ca566782 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:54:19 +0200
Subject: [PATCH 14/18] refactor: stop recomputing the DAG in sync submit_many

Sync submit_many normalized tasks, built the dependency graph, and ran a
topological sort only to discard the results, then called the async client
which computes all of it again. That was wasted work and a second, divergent
source of validation.

It now derives only the result keys it needs and delegates normalization,
dependency resolution, and ordering to the async client.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/client.py | 14 ++++----------
 1 file changed, 4 insertions(+), 10 deletions(-)

diff --git a/src/cashet/client.py b/src/cashet/client.py
index 6c83cda..c8d5e29 100644
--- a/src/cashet/client.py
+++ b/src/cashet/client.py
@@ -9,9 +9,6 @@
 from typing import Any, TypeVar, cast, overload
 
 from cashet._batch import (
-    build_deps,
-    normalize_tasks,
-    topological_sort,
     unpack_dict_tasks,
     unpack_list_tasks,
 )
@@ -193,15 +190,12 @@ def submit_many(
         max_workers: int | None = None,
     ) -> list[ResultRef[Any]] | dict[str, ResultRef[Any]]:
         is_dict = isinstance(tasks, dict)
+        # Only the keys are needed here to map results back; the async client owns
+        # normalization, dependency resolution, and topological ordering.
         if is_dict:
-            keys, raw_tasks = unpack_dict_tasks(tasks)
+            keys, _ = unpack_dict_tasks(tasks)
         else:
-            keys, raw_tasks = unpack_list_tasks(tasks)
-
-        key_set = set(keys)
-        normalized = normalize_tasks(raw_tasks, _cache, _tags, _retries, _force, _timeout, _ttl)
-        deps, _task_refs = build_deps(keys, normalized, key_set)
-        _order = topological_sort(deps)
+            keys, _ = unpack_list_tasks(tasks)
         workers = max_workers if max_workers is not None else self._max_workers
 
         async_results = self._runner.call(

From 5035c1e08c8131d11282355173c74665c6270e42 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 17:59:01 +0200
Subject: [PATCH 15/18] fix: cancel in-flight batch tasks when one fails

In parallel submit_many, a failing task re-raised straight out of the drain
loop while its sibling tasks kept running, so they were orphaned and could go
on writing commits after the error was already surfaced.

The drain loop now cancels the remaining tasks and awaits them before
propagating, so a batch failure stops its siblings cleanly.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/cashet/_batch.py | 29 ++++++++++++++-------
 tests/test_map.py    | 62 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 81 insertions(+), 10 deletions(-)

diff --git a/src/cashet/_batch.py b/src/cashet/_batch.py
index fff3398..9a1f790 100644
--- a/src/cashet/_batch.py
+++ b/src/cashet/_batch.py
@@ -228,16 +228,25 @@ def _schedule_ready() -> None:
             tasks[task] = key
 
     _schedule_ready()
-    while tasks:
-        done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED)
-        for task in done:
-            key = tasks.pop(task)
-            results[key] = task.result()
-            for dependent in rev[key]:
-                in_degree[dependent] -= 1
-                if in_degree[dependent] == 0:
-                    ready.append(dependent)
-            _schedule_ready()
+    try:
+        while tasks:
+            done, _ = await asyncio.wait(tasks.keys(), return_when=asyncio.FIRST_COMPLETED)
+            for task in done:
+                key = tasks.pop(task)
+                results[key] = task.result()
+                for dependent in rev[key]:
+                    in_degree[dependent] -= 1
+                    if in_degree[dependent] == 0:
+                        ready.append(dependent)
+                _schedule_ready()
+    except BaseException:
+        # On any failure, cancel and drain in-flight siblings so they cannot keep
+        # running (and writing commits) after the error has been surfaced.
+        for pending in tasks:
+            pending.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        raise
 
     logger.info("batch completed task_count=%d", len(keys))
     return results
diff --git a/tests/test_map.py b/tests/test_map.py
index 841e7c6..76c31c8 100644
--- a/tests/test_map.py
+++ b/tests/test_map.py
@@ -1,7 +1,10 @@
 from __future__ import annotations
 
+import asyncio
 from pathlib import Path
+from typing import Any
 
+import pytest
 import pytest_asyncio
 
 from cashet import Client
@@ -93,3 +96,62 @@ def counter(x: int) -> int:
     async def test_map_empty(self, async_client: AsyncClient) -> None:
         refs = await async_client.map(double, [])
         assert refs == []
+
+
+class TestBatchCancellation:
+    async def test_failure_cancels_in_flight_siblings(self) -> None:
+        from cashet._batch import execute_batch
+        from cashet.models import Commit, TaskDef, TaskError, TaskStatus
+
+        sibling_started = asyncio.Event()
+        sibling_cancelled = asyncio.Event()
+
+        def failing() -> int:
+            return 0
+
+        def sibling() -> int:
+            return 0
+
+        fail_commit = Commit(
+            hash="f" * 64,
+            task_def=TaskDef(
+                func_hash="x", func_name="failing", func_source="",
+                args_hash="y", args_snapshot=b"",
+            ),
+            output_ref=None,
+            error="boom",
+            status=TaskStatus.FAILED,
+        )
+
+        class StubExecutor:
+            async def submit(self, func: Any, *_: Any) -> tuple[Commit, bool]:
+                if func is sibling:
+                    sibling_started.set()
+                    try:
+                        await asyncio.sleep(30)
+                    except asyncio.CancelledError:
+                        sibling_cancelled.set()
+                        raise
+                    raise AssertionError("sibling should have been cancelled")
+                await sibling_started.wait()
+                return fail_commit, False
+
+        keys: list[Any] = ["fail", "sib"]
+        normalized: list[Any] = [
+            (failing, (), {}, True, {}, 0, False, None, None),
+            (sibling, (), {}, True, {}, 0, False, None, None),
+        ]
+        with pytest.raises(TaskError):
+            await execute_batch(
+                order=keys,
+                keys=keys,
+                normalized=normalized,
+                task_refs={},
+                deps={"fail": set(), "sib": set()},
+                executor=StubExecutor(),
+                store=None,  # type: ignore[arg-type]
+                serializer=None,  # type: ignore[arg-type]
+                max_workers=2,
+            )
+        assert sibling_started.is_set()
+        assert sibling_cancelled.is_set()

From aad36094ae05e7823dcd36205ab9324bb913922a Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 18:00:18 +0200
Subject: [PATCH 16/18] chore: release 0.4.5

Bump version to 0.4.5 and document the correctness, robustness, server
hardening, and performance fixes in the changelog.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md   | 55 ++++++++++++++++++++++++++++++++++++++++++++++++++
 pyproject.toml |  2 +-
 uv.lock        |  2 +-
 3 files changed, 57 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index d20d4da..c02a463 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,60 @@
 # Changelog
 
+## 0.4.5 - 21.6.2026.
+
+### Fixed
+- Hash objects by their `__slots__`/`__dict__` state instead of the memory
+  address. Slotted value types (including `@dataclass(slots=True)`) cached by
+  identity before, so identical calls missed the cache every run and lossy
+  custom reprs could collide. Truly opaque objects now warn instead of silently
+  address-hashing.
+- Record `ResultRef` values nested inside custom-object attributes in commit
+  lineage, matching what the hasher already accounts for, so garbage collection
+  cannot evict a blob a commit still depends on.
+- Hash referenced module-global `dict`/`list`/`set` values consistently with
+  scalars, so config-style globals invalidate the cache when they change.
+- Canonicalize function source with `ast.unparse` instead of `ast.dump`, which
+  keeps the same comment/whitespace/docstring insensitivity but stays stable
+  across Python versions whose `ast.dump` field set differs.
+- Treat the `last_accessed_at` bump on a cache hit as best-effort so a locked
+  database no longer turns a successful lookup into an error.
+- Skip a post-eviction `VACUUM` that is blocked by a concurrent writer instead
+  of failing an eviction whose deletes already committed.
+- Prevent Redis tag index key collisions: presence and value indexes now use
+  distinct prefixes and the value key length-prefixes the tag key, so a `:` in
+  a key or value can no longer cross-match unrelated commits.
+- Make the Redis blob delete script idempotent against a missing reference
+  counter so it cannot drop a blob another commit still references.
+- Enforce the HTTP server request size limit on the bytes actually received,
+  closing a bypass for chunked requests that omit `Content-Length`.
+- Validate handler input (returning 400) and wrap every server handler in a
+  generic 500 barrier so malformed input and internal errors no longer leak
+  stack traces; `/gc` also accepts an empty body and falls back to defaults.
+- Report commits skipped during a lossy import. A truncated archive no longer
+  looks like a clean import.
+- Exit non-zero from `cashet show` and `cashet get` when a commit is missing,
+  matching `cashet rm`.
+- Cancel in-flight sibling tasks when a parallel `submit_many` task fails,
+  instead of leaving them running after the error is surfaced.
+
+### Performance
+- Redis `find_running_by_fingerprint` uses a per-fingerprint running-claim index
+  for O(1) lookup instead of scanning a fingerprint's full commit history on
+  every submit.
+
+### Changed
+- `import_archive` (sync and async) now returns an `ImportResult(imported,
+  skipped)` named tuple instead of a bare imported count.
+- Sync `submit_many` no longer recomputes the dependency graph that the async
+  client already builds.
+
+### Notes
+- The hashing fixes change function and argument cache keys, so results cached
+  by earlier versions recompute on first access. Blobs are content-addressed and
+  remain until garbage collected.
+- The Redis tag index key scheme changed. Tag indexes written by older versions
+  are not migrated; rewrite affected commits to rebuild them.
+
 ## 0.4.4 — 11.5.2026.
 
 ### Fixed
diff --git a/pyproject.toml b/pyproject.toml
index 1819618..36692e5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "cashet"
-version = "0.4.4"
+version = "0.4.5"
 description = "A Python memoization cache with Redis, async support, DAG pipelines, and an HTTP server"
 readme = "README.md"
 license = "MIT"
diff --git a/uv.lock b/uv.lock
index c084fac..8f80400 100644
--- a/uv.lock
+++ b/uv.lock
@@ -26,7 +26,7 @@ wheels = [
 
 [[package]]
 name = "cashet"
-version = "0.4.4"
+version = "0.4.5"
 source = { editable = "." }
 dependencies = [
     { name = "click" },

From 5c5512d3eb969b1be78fd177412b80ada2034111 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 18:04:16 +0200
Subject: [PATCH 17/18] docs: rewrite README for clarity and concision

Cut the README roughly in half by removing duplicated sections (the HTTP
server was documented three times), tightening prose, and consolidating the
per-task option and serialization docs into single references.

Fix the import_archive example for the new ImportResult return value and add a
0.4.4 to 0.4.5 compatibility note.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 README.md | 894 ++++++++++++------------------------------------------
 1 file changed, 202 insertions(+), 692 deletions(-)

diff --git a/README.md b/README.md
index d18162d..f89050d 100644
--- a/README.md
+++ b/README.md
@@ -1,18 +1,22 @@
 <h1 align="center">cashet</h1>
 
 <p align="center">
-  <strong>A Python memoization cache with Redis, async support, and an HTTP server.</strong><br>
-  Hash functions + args into cache keys. Results stored as immutable blobs. Chain outputs with DAG resolution.<br>
-  Run a function once — get the same result instantly every time after.
+  <strong>Content-addressable compute cache for Python: persistent function memoization with Redis, async, DAG pipelines, and an HTTP server.</strong><br>
+  Hash a function plus its arguments into a cache key, store the result as an immutable blob, and return it instantly on every later call.<br>
+  Think git for your function results.
+</p>
+
+<p align="center">
+  <em>Keywords: Python caching, memoization, disk cache, Redis cache, async asyncio, DAG pipeline, content-addressable, reproducible computation, joblib / lru_cache alternative.</em>
 </p>
 
 <p align="center">
   <a href="#install">Install</a> ·
-  <a href="#quickstart">Quick Start</a> ·
-  <a href="#why">Why</a> ·
+  <a href="#quick-start">Quick Start</a> ·
+  <a href="#why-cashet">Why cashet</a> ·
   <a href="#use-cases">Use Cases</a> ·
   <a href="#cli">CLI</a> ·
-  <a href="#api">API</a> ·
+  <a href="#python-api">Python API</a> ·
   <a href="#how-it-works">How It Works</a>
 </p>
 
@@ -24,9 +28,15 @@
 
 ---
 
+## What is cashet
+
+cashet is a caching library for expensive Python functions. You submit a function and its arguments, cashet runs it once, stores the result as a content-addressed blob, and serves that result instantly on every later call with the same code and inputs. Caches persist across process restarts, can be shared across machines through Redis, and can be inspected, diffed, and chained from the CLI or the Python API.
+
+The cache key is a SHA-256 hash of the function's AST-normalized source plus its arguments. Comments, docstrings, and formatting do not invalidate the cache; only a real change to the code or the inputs does. Identical results are deduplicated to a single blob on disk, and every result is a git-like object you can inspect and chain into pipelines.
+
 ## Install
 
-**Global CLI tool** (recommended):
+Install as a global CLI tool:
 
 ```bash
 uv tool install cashet
@@ -34,13 +44,7 @@ uv tool install cashet
 pipx install cashet
 ```
 
-Then use the CLI anywhere:
-
-```bash
-cashet --help
-```
-
-**In a project** (library + CLI):
+Add to a project as a library and project-local CLI:
 
 ```bash
 uv add cashet
@@ -48,31 +52,15 @@ uv add cashet
 pip install cashet
 ```
 
-This installs `cashet` as both an importable Python library (`from cashet import Client`) and a project-local CLI (`uv run cashet`).
-
-**With Redis backend:**
-
-```bash
-uv add "cashet[redis]"
-# or
-pip install "cashet[redis]"
-```
-
-**With HTTP server:**
-
-```bash
-uv add "cashet[server]"
-# or
-pip install "cashet[server]"
-```
-
-**All extras:**
+Optional backends:
 
 ```bash
+uv add "cashet[redis]"     # shared cache via Redis
+uv add "cashet[server]"    # HTTP server
 uv add "cashet[redis,server]"
 ```
 
-**Develop / contribute:**
+Develop and contribute:
 
 ```bash
 git clone https://github.com/jolovicdev/cashet.git
@@ -81,37 +69,27 @@ uv sync --all-extras
 uv run pytest
 ```
 
-## Version Compatibility
-
-**0.2.x → 0.3.x: Hash format changed.** `hash_args` was unified with `serialize_args` to use a single canonical representation via `_stable_repr_to`. Same function + same arguments produce different cache fingerprints across the major version line. Caches created with 0.2.x **will not hit** on 0.3.x.
-
-**0.3.0 → 0.3.1:** Redis blob data keys renamed from `cashet:blob:{hash}` to `cashet:blob:data:{hash}` to fix a stats double-counting bug. Redis blob object/byte totals are maintained in `cashet:stats:blob` after a one-time backfill for existing caches. Function hashing also now includes defaults / keyword defaults in more cases, and custom object argument hashing includes the defining module. Existing 0.3.0 caches may miss after upgrading. Clear old caches before upgrading if you use Redis or rely on long-lived cache reuse across versions.
-
-**0.3.x → 0.4.0:** Added per-entry TTL and tag-based invalidation. SQLite stores auto-migrate (new `ttl_seconds` and `expires_at` columns). Redis stores encode new fields in commit JSON. Old caches are fully readable after upgrading.
-
-**Upgrading from an incompatible version:** either clear the cache (`cashet clear`) and rebuild, or point the new version at a fresh store directory.
-
 ## Quick Start
 
 ```python
 from cashet import Client
 
-client = Client()  # creates .cashet/ in current directory
+client = Client()  # creates .cashet/ in the current directory
 
 def expensive_transform(data, scale=1.0):
     # imagine this takes 10 minutes
     return [x * scale for x in data]
 
-# First call: runs the function
+# First call runs the function
 ref = client.submit(expensive_transform, [1, 2, 3], scale=2.0)
 print(ref.load())  # [2.0, 4.0, 6.0]
 
-# Second call with same args: instant — returns cached result
+# Same args again: instant, returns the cached result with no re-computation
 ref2 = client.submit(expensive_transform, [1, 2, 3], scale=2.0)
-print(ref2.load())  # [2.0, 4.0, 6.0] — no re-computation
+print(ref2.load())  # [2.0, 4.0, 6.0]
 ```
 
-You can also use `Client` as a context manager to ensure the store connection is closed cleanly:
+`Client` works as a context manager so the store connection closes cleanly:
 
 ```python
 with Client() as client:
@@ -119,50 +97,37 @@ with Client() as client:
     print(ref.load())
 ```
 
-Chain tasks into a pipeline where each step's output feeds into the next:
+### Pipelines
 
-```python
-from cashet import Client
+Pass a result reference straight into the next task. Each step's output feeds the next, and cashet records the lineage:
 
+```python
 client = Client()
 
-def load_dataset(path):
-    return list(range(100))
-
-def normalize(data):
-    max_val = max(data)
-    return [x / max_val for x in data]
-
-def train_model(data, lr=0.01):
-    return {"loss": 0.05, "lr": lr, "samples": len(data)}
-
-# Step 1: load
 raw = client.submit(load_dataset, "data/train.csv")
-
-# Step 2: normalize (receives raw output as input)
-normalized = client.submit(normalize, raw)
-
-# Step 3: train (receives normalized output)
+normalized = client.submit(normalize, raw)            # receives raw's output
 model = client.submit(train_model, normalized, lr=0.001)
 
-print(model.load())  # {'loss': 0.05, 'lr': 0.001, 'samples': 100}
+print(model.load())
 ```
 
-Re-run the script — everything returns instantly from cache. Change one argument and only that step (and downstream) re-runs.
+Re-run the script and everything returns instantly from cache. Change one argument and only that step and the steps downstream of it re-run.
 
-**Shared cache with Redis** — multiple processes or machines share one cache:
+### Shared cache with Redis
+
+Multiple processes or machines share one cache:
 
 ```python
 from cashet import Client
 from cashet.redis_store import RedisStore
 
-client = Client(store=RedisStore("redis://localhost:6379"))
-
-# Any process on any machine using the same Redis gets cached results
+client = Client(store=RedisStore("redis://localhost:6379/0"))
 ref = client.submit(expensive_transform, [1, 2, 3], scale=2.0)
 ```
 
-**Async API** — drop-in `AsyncClient` for asyncio workflows:
+### Async
+
+`AsyncClient` is a drop-in for asyncio workflows and accepts both sync and async callables:
 
 ```python
 import asyncio
@@ -170,31 +135,16 @@ from cashet.async_client import AsyncClient
 
 async def main():
     client = AsyncClient()
-
-    def compute(x: int) -> int:
-        return x * 2
-
-    ref = await client.submit(compute, 21)
-    result = await ref.load()
-    print(result)  # 42
+    ref = await client.submit(lambda x: x * 2, 21)
+    print(await ref.load())  # 42
+    await client.close()
 
 asyncio.run(main())
 ```
 
-**HTTP server** — expose cache metadata and registered tasks over HTTP:
-
-```bash
-python -c "from cashet import Client; Client().serve(port=8000)"
-```
-
-```python
-# With auth token
-from cashet import Client
-client = Client()
-client.serve(port=8000, require_token="my-secret-token")
-```
+### HTTP server
 
-Register server-side tasks in the server process:
+Expose cache metadata and registered tasks over HTTP:
 
 ```python
 from cashet import Client
@@ -207,26 +157,19 @@ def double(x):
 client.serve(port=8000, tasks={"double": double})
 ```
 
-Submit JSON args from another process:
-
 ```python
 import requests
 
-r = requests.post("http://localhost:8000/submit", json={
-    "task": "double",
-    "args": [5],
-})
+requests.post("http://localhost:8000/submit", json={"task": "double", "args": [5]})
 ```
 
-Submitting Python source, dill payloads, or serializer-encoded args is disabled by default.
-For trusted internal clients only, enable the legacy remote-code path with both
-`allow_remote_code=True` and a non-empty `require_token=...`.
+By default the server only runs tasks registered in the server process. Submitting Python source, dill payloads, or serializer-encoded args is disabled unless you explicitly opt in (see [HTTP server endpoints](#http-server-endpoints)).
 
-## Why
+## Why cashet
 
-You already have caches (`functools.lru_cache`, `joblib.Memory`). Here's what's different:
+You already have `functools.lru_cache` and `joblib.Memory`. cashet adds persistence, sharing, provenance, and pipelines on top of memoization:
 
-| | lru_cache | joblib.Memory | **cashet** |
+| | lru_cache | joblib.Memory | cashet |
 |---|---|---|---|
 | AST-normalized hashing | No | No | Yes |
 | DAG / pipeline chaining | No | No | Yes |
@@ -241,359 +184,112 @@ You already have caches (`functools.lru_cache`, `joblib.Memory`). Here's what's
 | HTTP server | No | No | Yes |
 | Persists across restarts | No | Yes | Yes |
 
-The core idea: **hash the function's AST-normalized source + arguments = unique cache key**. Comments, docstrings, and formatting changes don't invalidate the cache — only semantic changes do. Same function + same args = same result, stored immutably on disk. The result is a git-like blob you can inspect, diff, and chain.
+The core idea: hash the function's AST-normalized source plus its arguments into a unique cache key. Same function and same args give the same result, stored immutably on disk, that you can inspect, diff, and chain.
 
 ## Use Cases
 
-### 1. ML Experiment Tracking Without the Bloat
-
-You run 200 hyperparameter sweeps overnight. Half crash. You fix a bug and re-run. Without cashet, you re-process the dataset 200 times. With cashet:
+**ML experiment tracking.** Run many hyperparameter sweeps that share one expensive preprocessing step. `preprocess` runs once and every training job reuses its cached output. Use `TaskRef(index)` to wire one task's output into another within a batch:
 
 ```python
-from cashet import Client, TaskError, TaskRef
+from cashet import Client, TaskRef
 
 client = Client()
 
-def preprocess(dataset_path, image_size):
-    # 45 minutes of image resizing
-    ...
-
-def train(data, learning_rate, dropout):
-    ...
-
-# Batch submit with topological ordering
-# TaskRef(0) refers to the first task's output
 results = client.submit_many([
-    (preprocess, ("s3://my-bucket/images", 224)),
+    (preprocess, ("s3://bucket/images", 224)),
     (train, (TaskRef(0), 0.01, 0.2)),
-    (train, (TaskRef(0), 0.01, 0.5)),
     (train, (TaskRef(0), 0.001, 0.2)),
-    (train, (TaskRef(0), 0.001, 0.5)),
     (train, (TaskRef(0), 0.0001, 0.2)),
-    (train, (TaskRef(0), 0.0001, 0.5)),
 ])
 ```
 
-`preprocess` runs **once** — all 6 training jobs reuse its cached output. Re-run the script tomorrow and even the training results come from cache (same function + same args = instant).
-
-### 2. Data Pipeline Debugging
-
-Your ETL pipeline fails at step 5. You fix a typo. Now you need to re-run steps 5-7 but steps 1-4 are unchanged and expensive:
-
-```python
-from cashet import Client
-
-client = Client()
-
-raw = client.submit(load_s3, "s3://logs/2024-05-01/")
-clean = client.submit(remove_pii, raw)
-enriched = client.submit(join_crm, clean, "select * from users")
-report = client.submit(generate_report, enriched)
-```
-
-Fix the `join_crm` function and re-run the script. Steps 1-2 return instantly from cache. Only step 3 onward re-executes. This works because cashet tracks which function produced which output — changing a function's source code changes its hash, invalidating downstream cache entries.
-
-### 3. Reproducible Notebook Results
-
-`cashet` is designed to work in Jupyter notebooks and IPython sessions. Share a result with a colleague and they can verify exactly how it was produced:
-
-```python
-# your notebook
-ref = client.submit(generate_forecast, date="2024-01-01", model="v3")
-print(f"Result hash: {ref.hash}")
-```
-
-```bash
-# their terminal — inspect provenance
-cashet show <hash>
-
-# Output:
-# Hash:     a3b4c5d6...
-# Function: generate_forecast
-# Source:   def generate_forecast(date, model): ...
-# Args:     (('2024-01-01',), {'model': 'v3'})
-# Created:  2024-05-01T10:32:17
-
-# Retrieve the actual result
-cashet get <hash> -o forecast.csv
-```
+**Data pipeline debugging.** Your ETL job fails at step 5. Fix the function and re-run the script. Unchanged upstream steps return from cache; only the changed step and everything after it re-executes, because changing a function's source changes its hash and invalidates downstream entries.
 
-### 4. Incremental Computation
+**Reproducible notebooks.** Share a result hash and a colleague can inspect exactly how it was produced from their terminal with `cashet show <hash>` and retrieve it with `cashet get <hash> -o out.bin`.
 
-Process a large dataset in chunks. Already-processed chunks return instantly:
+**Incremental computation.** Process a large dataset in chunks with `client.map`. Already-processed chunks return instantly; add a new chunk and only that one runs:
 
 ```python
-from cashet import Client
-
-client = Client()
-
-def process_chunk(chunk_id, source_file):
-    # expensive per-chunk processing
-    ...
-
-results = []
-for chunk_id in range(100):
-    ref = client.submit(process_chunk, chunk_id, "huge_file.parquet")
-    results.append(ref)
+refs = client.map(process_chunk, range(100), source_file="data.parquet")
+results = [r.load() for r in refs]
 ```
 
-First run processes all 100 chunks. Second run (even after restarting Python) returns all 100 results instantly. Add a new chunk? Only that one runs.
-
 ## CLI
 
 ```bash
-# Show commit history
-cashet log
-
-# Filter by function name
-cashet log --func "preprocess"
-
-# Filter by tag
-cashet log --tag env=prod --tag experiment=run-1
-
-# Show full commit details (source code, args, error)
-cashet show <hash>
-
-# Retrieve a result (pretty-prints strings/dicts/lists)
-cashet get <hash>
-
-# Write a result to file
-cashet get <hash> -o output.bin
-
-# Compare two commits
-cashet diff <hash_a> <hash_b>
-
-# Show lineage of a result (same function+args over time)
-cashet history <hash>
-
-# Delete a specific commit
-cashet rm <hash>
-
-# Evict old cache entries and orphaned blobs
-cashet gc --older-than 30
-
-# Evict oldest entries until under a size limit
-cashet gc --max-size 1GB
-
-# Clear everything (alias for gc --older-than 0)
-cashet clear
-
-# Export all commits and blobs to a tar.gz archive
-cashet export backup.tar.gz
-
-# Import commits and blobs from a tar.gz archive
-cashet import backup.tar.gz
-
-# Storage statistics (includes disk size)
-cashet stats
-
-# Start the HTTP server
+cashet log                         # commit history
+cashet log --func preprocess       # filter by function
+cashet log --tag env=prod          # filter by tag
+cashet show <hash>                 # full commit details (source, args, error)
+cashet get <hash>                  # retrieve a result (pretty-prints str/dict/list)
+cashet get <hash> -o output.bin    # write a result to a file
+cashet diff <hash_a> <hash_b>      # compare two commits
+cashet history <hash>              # lineage of one function+args over time
+cashet rm <hash>                   # delete a commit
+cashet invalidate -t env=prod      # delete commits by tag
+cashet gc --older-than 30          # evict entries older than 30 days
+cashet gc --max-size 1GB           # evict oldest entries until under a size limit
+cashet clear                       # remove everything
+cashet export backup.tar.gz        # export commits and blobs to an archive
+cashet import backup.tar.gz        # import from an archive
+cashet stats                       # storage statistics
 cashet serve --host 127.0.0.1 --port 8000
-
-# Legacy unsafe remote-code mode requires auth
-cashet serve --require-token secret123 --allow-remote-code
-```
-
-## API
-
-### `Client`
-
-```python
-from cashet import Client
-
-client = Client(
-    store_dir=".cashet",       # where to store blobs + metadata (SQLiteStore)
-                               # falls back to $CASHET_DIR env var if set
-    store=None,                # or inject any Store implementation (SQLiteStore, RedisStore)
-    executor=None,             # or inject any Executor implementation
-    serializer=None,           # defaults to PickleSerializer
-    max_workers=1,             # max parallelism for submit_many (default: 1, sequential)
-)
 ```
 
-### `AsyncClient`
-
-```python
-import asyncio
-from cashet.async_client import AsyncClient
-
-async def main():
-    client = AsyncClient(
-        store_dir=".cashet",   # defaults to AsyncSQLiteStore
-        store=None,            # or AsyncRedisStore, or any AsyncStore
-        executor=None,         # defaults to AsyncLocalExecutor
-        serializer=None,       # defaults to PickleSerializer
-        max_workers=1,         # max parallelism for submit_many (default: 1, sequential)
-    )
-
-    async def square(x: int) -> int:
-        return x * x
-
-    ref = await client.submit(square, 5)
-    result = await ref.load()  # 25
-    await client.close()
-
-asyncio.run(main())
-```
-
-`AsyncClient` mirrors `Client` — `submit()`, `submit_many()`, `log()`, `show()`, `get()`, `stats()`, `gc()`, `rm()`, `clear()`, `serve()` are all `async def`. `submit()` accepts both sync and async callables, and returns `AsyncResultRef` with `async load()`. Chain tasks by passing `AsyncResultRef` as an argument.
-
-### HTTP Server
-
-```python
-# Start the server
-from cashet import Client
-client = Client()
-client.serve(host="127.0.0.1", port=8000)
-
-# With Bearer token authentication
-client.serve(host="0.0.0.0", port=8000, require_token="secret123")
-```
-
-Execute tasks over HTTP by registering callables in the server process and sending
-JSON arguments from another process:
-
-```python
-from cashet import Client
-
-client = Client()
-
-def add(x: int, y: int) -> int:
-    return x + y
-
-client.serve(port=8000, tasks={"add": add})
-```
-
-```python
-import requests
-
-r = requests.post("http://localhost:8000/submit", json={"task": "add", "args": [3, 4]})
-```
-
-Endpoints:
-
-| Method | Path | Description |
-|---|---|---|
-| POST | `/submit` | Submit a registered task for execution |
-| GET | `/result/{hash}` | Fetch deserialized result |
-| GET | `/commit/{hash}` | Commit metadata |
-| GET | `/log` | List commits (query: `?func=`, `?limit=`, `?status=`) |
-| GET | `/stats` | Storage statistics |
-| POST | `/gc` | Run garbage collection |
-
-Use `AsyncClient.serve()` for the async variant with `create_async_app()`.
+`cashet show`, `cashet get`, and `cashet rm` exit with a non-zero status when the commit is not found, so they compose in scripts.
 
-> **Security:** `/submit` does not execute client-supplied Python by default. The legacy
-> `func_source`, `func_b64`, `args_b64`, and `kwargs_b64` payloads require
-> `allow_remote_code=True` and a non-empty Bearer token. In the CLI, this is
-> `cashet serve --require-token secret123 --allow-remote-code`. That mode deserializes
-> and executes arbitrary Python, so expose it only to trusted clients.
+## Python API
 
-### Redis Backend
+### Client and AsyncClient
 
 ```python
 from cashet import Client
-from cashet.redis_store import RedisStore
-
-# Sync
-client = Client(store=RedisStore("redis://localhost:6379/0"))
-
-# Async
-from cashet.async_client import AsyncClient
-from cashet.redis_store import AsyncRedisStore
-
-client = AsyncClient(store=AsyncRedisStore("redis://localhost:6379/0"))
-```
-
-Redis-backed stores support all operations — cache dedup, fingerprint lookup, history, eviction, and size-based GC. Cross-process claim dedup uses per-fingerprint Redis locks (`cashet:lock:{fingerprint}`) to prevent redundant execution across machines. Last-access eviction is indexed in Redis (`cashet:index:last_accessed`), blob stats are maintained in `cashet:stats:blob`, and blob ref counts (`cashet:blob:ref:{hash}`) enable orphan cleanup without scanning all commits.
 
-### Pluggable Backends
-
-Everything is protocol-based. Swap the store, executor, or serializer without touching your task code:
-
-```python
-from pathlib import Path
-
-from cashet import Client, Store, Executor, Serializer
-from cashet.store import SQLiteStore
-from cashet.executor import LocalExecutor
-
-# These are equivalent (the defaults):
-client = Client(store_dir=".cashet")
-
-# Explicit injection:
 client = Client(
-    store=SQLiteStore(Path(".cashet")),
-    executor=LocalExecutor(),
+    store_dir=".cashet",   # blob + metadata directory; falls back to $CASHET_DIR
+    store=None,            # or any Store (SQLiteStore, RedisStore, ...)
+    executor=None,         # or any Executor
+    serializer=None,       # defaults to PickleSerializer
+    max_workers=1,         # parallelism for submit_many (1 = sequential)
 )
 ```
 
-**Store protocol** — implement this to use RocksDB, Redis, S3, or anything else:
+`AsyncClient` (from `cashet.async_client`) mirrors `Client`: `submit`, `submit_many`, `map`, `log`, `show`, `get`, `diff`, `history`, `stats`, `gc`, `rm`, `clear`, `invalidate`, and `serve` are all `async def`. It returns `AsyncResultRef` with an `async load()`.
 
-```python
-from cashet.protocols import Store
-
-class RedisStore:
-    def put_blob(self, data: bytes) -> ObjectRef: ...
-    def get_blob(self, ref: ObjectRef) -> bytes: ...
-    def put_commit(self, commit: Commit) -> None: ...
-    def get_commit(self, hash: str) -> Commit | None: ...
-    def find_by_fingerprint(self, fingerprint: str) -> Commit | None: ...
-    def find_running_by_fingerprint(self, fingerprint: str) -> Commit | None: ...
-    def list_commits(self, ...) -> list[Commit]: ...
-    def get_history(self, hash: str) -> list[Commit]: ...
-    def stats(self) -> dict[str, int]: ...
-    def evict(self, older_than: datetime, max_size_bytes: int | None = None) -> int: ...
-    def delete_commit(self, hash: str) -> bool: ...
-    def close(self) -> None: ...
-
-client = Client(store=RedisStore("redis://localhost"))
-# Everything else works identically
-```
-
-**Executor protocol** — implement this for distributed execution (Celery, Kafka, RQ):
+### submit and ResultRef
 
 ```python
-from cashet.protocols import Executor
-
-class CeleryExecutor:
-    def submit(self, func, args, kwargs, task_def, store, serializer):
-        # Push to Celery, poll for result
-        ...
-
-client = Client(
-    store=RedisStore("redis://localhost"),
-    executor=CeleryExecutor(),
-)
+ref = client.submit(my_func, arg1, arg2, key="value")
+ref.hash          # content hash of the result blob
+ref.commit_hash   # commit hash (use for show / history / rm / get)
+ref.size          # size in bytes
+ref.load()        # deserialize and return the result
 ```
 
-**Serializer protocol** — already covered below.
+If the same function and arguments were submitted before, `submit` returns the cached result without re-executing. `ResultRef` is generic, so `submit` infers the return type from the function's annotation. Pass a `ResultRef` as an argument (including nested inside lists, tuples, sets, frozensets, and dicts) to chain tasks; it resolves to its output before execution.
 
-### `client.submit(func, *args, **kwargs) -> ResultRef`
+### Per-task options
 
-Submit a function for execution. Returns a `ResultRef` — a lazy handle to the result.
+Each option is available per call (prefixed with `_`) and on the `@client.task` decorator:
 
 ```python
-ref = client.submit(my_func, arg1, arg2, key="value")
-ref.hash         # content hash of the result blob
-ref.commit_hash  # commit hash (use this for show/history/rm/get)
-ref.size         # size in bytes
-ref.load()       # deserialize and return the result
-```
+ref = client.submit(fetch_api, url, _retries=3, _timeout=30, _ttl=3600, _tags={"env": "prod"})
 
-If the same function + same arguments have been submitted before, returns the cached result **without re-executing**.
-
-### `client.clear()`
-
-Remove all cache entries and orphaned blobs. Equivalent to `client.gc(timedelta(days=0))`.
+@client.task(cache=False, retries=3, timeout=30, ttl=3600, tags={"team": "ml"})
+def fetch_api(url):
+    ...
 
-```python
-client.clear()
+ref = my_task(url)  # decorated tasks are directly callable and return a ResultRef
 ```
 
-### `client.submit_many(tasks) -> list[ResultRef]`
+- `cache=False` runs the function every time but still records lineage (non-deterministic work).
+- `force=True` skips the cache and re-executes once.
+- `retries=N` retries failed attempts, then raises `TaskError` with the original traceback.
+- `timeout=seconds` bounds an attempt. Local timeouts are soft: cashet stops waiting and records a failure, but Python cannot kill running thread code, so keep task functions idempotent.
+- `ttl=seconds` expires a commit. Expired commits are skipped on lookup and removed at garbage collection.
+- `tags={...}` attach metadata for filtering. Tags are not part of the cache key.
 
-Submit a batch of tasks with automatic topological ordering. Use `TaskRef(index)` to wire outputs between tasks in the batch.
+### submit_many and map
 
 ```python
 from cashet import TaskRef
@@ -601,367 +297,181 @@ from cashet import TaskRef
 refs = client.submit_many([
     step1_func,
     (step2_func, (TaskRef(0),)),
-    (step3_func, (TaskRef(1), "extra_arg")),
-], max_workers=4)  # run independent tasks in parallel
-```
-
-This enables parallel fan-out and ensures each task only runs after its dependencies.
-
-> **Note:** With the default `SQLiteStore`, parallel execution serializes on the SQLite write lock — `max_workers > 1` only benefits compute-heavy tasks where execution time dominates. For true parallel fan-out, use `RedisStore`.
+    (step3_func, (TaskRef(1), "extra")),
+], max_workers=4)
 
-### `client.map(func, items, *args, **kwargs) -> list[ResultRef]`
-
-Map a function over an iterable with per-item caching. Each item becomes the first positional argument; additional `*args` and `**kwargs` are appended.
-
-```python
 refs = client.map(process_chunk, range(100), source_file="data.parquet")
-# refs[0].load() == process_chunk(0, source_file="data.parquet")
-
-results = [r.load() for r in refs]
-```
-
-Already-computed items return instantly from cache. Add a new chunk later and only that item re-runs.
-
-**Opt out of caching:**
-
-```python
-# Per-call
-ref = client.submit(non_deterministic_func, _cache=False)
-
-# Per-function via decorator
-@client.task(cache=False)
-def random_score():
-    return random.random()
-```
-
-**Force re-execution (skip cache, always run):**
-
-```python
-# Per-call
-ref = client.submit(my_func, arg, _force=True)
-
-# Per-function via decorator
-@client.task(force=True)
-def always_rerun():
-    ...
-```
-
-**Tag commits:**
-
-```python
-# Per-call
-ref = client.submit(train, data, lr=0.01, _tags={"experiment": "v1"})
-
-# Per-function via decorator
-@client.task(tags={"team": "ml"})
-def preprocess(raw):
-    ...
-```
-
-Tags are not part of the cache key — they are metadata for organization and filtering.
-
-**Retry flaky operations:**
-
-```python
-# Per-call
-ref = client.submit(fetch_api, url, _retries=3)
-
-# Per-function via decorator
-@client.task(retries=3)
-def fetch_api(url):
-    ...
-```
-
-Retries wait briefly between attempts. When retries are exhausted, `client.submit` raises `TaskError` with the original traceback included in the message.
-
-**Task timeouts:**
-
-```python
-# Per-call (seconds)
-ref = client.submit(slow_func, _timeout=30)
-
-# Per-function via decorator
-@client.task(timeout=30)
-def slow_func():
-    ...
-```
-
-Timeouts can be combined with retries — a timed-out attempt counts as a failure and will be retried. Local execution uses Python threads, so timeouts are soft: cashet stops waiting and records the attempt as failed, but Python cannot safely kill already-running thread code. Use idempotent task functions; hard cancellation belongs in a process/distributed executor such as a future Celery executor.
-
-**Per-entry TTL:**
-
-```python
-# Per-call (seconds)
-ref = client.submit(fetch_api_data, url, _ttl=3600)
-
-# Per-function via decorator
-@client.task(ttl=3600)
-def fetch_api_data(url):
-    ...
 ```
 
-Commits expire automatically after the TTL. Expired commits are skipped by cache lookups and history queries but are not physically deleted until garbage collection runs.
-
-### `@client.task`
+`submit_many` orders tasks topologically and runs independent ones in parallel up to `max_workers`. With the default `SQLiteStore`, parallel execution serializes on the SQLite write lock, so `max_workers > 1` mainly helps compute-heavy tasks; use `RedisStore` for true fan-out across processes.
 
-Register a function with cashet metadata and make it directly callable:
+### Inspecting, diffing, and eviction
 
 ```python
-@client.task
-def my_func(x):
-    return x * 2
-
-ref = my_func(5)  # Returns ResultRef, same as client.submit(my_func, 5)
-ref.load()        # 10
-
-@client.task(cache=False, name="custom_task_name", tags={"env": "prod"})
-def other_func(x):
-    return x + 1
-```
-
-`client.submit(my_func, 5)` still works identically.
-
-### `client.log()`, `client.show()`, `client.get()`, `client.diff()`, `client.history()`, `client.rm()`, `client.gc()`
-
-```python
-# List commits
-commits = client.log(func_name="preprocess", limit=10)
-
-# Filter by status
-commits = client.log(status="failed")
+commits = client.log(func_name="preprocess", limit=10, status="failed", tags={"experiment": "v1"})
 
-# Filter by tags
-commits = client.log(tags={"experiment": "v1"})
-
-# Get commit details
 commit = client.show(hash)
-commit.task_def.func_source  # the source code
-commit.task_def.args_snapshot  # the serialized args
-commit.parent_hash  # previous commit for same func+args
-commit.created_at
+commit.task_def.func_source     # the source code
+commit.task_def.args_snapshot   # the serialized args
+commit.parent_hash              # previous commit for the same func + args
 
-# Load a result by commit hash
 result = client.get(hash)
+diff = client.diff(hash_a, hash_b)   # {'func_changed': ..., 'args_changed': ..., 'output_changed': ...}
+history = client.history(hash)       # all runs of the same func + args
 
-# Diff two commits
-diff = client.diff(hash_a, hash_b)
-# {'func_changed': True, 'args_changed': False, 'output_changed': True, ...}
-
-# Get lineage (all runs of same func+args)
-history = client.history(hash)
-
-# Evict old entries (default: 30 days)
-evicted = client.gc()
-# Evict entries older than 7 days
 from datetime import timedelta
-evicted = client.gc(older_than=timedelta(days=7))
-# Evict oldest entries until under size limit
-evicted = client.gc(max_size_bytes=1024 * 1024 * 1024)  # 1GB
-
-# Invalidate commits by tags
-deleted = client.invalidate(tags={"experiment": "v1"})
-# deleted = client.invalidate(tags={"experiment": None})  # any commit with 'experiment' tag
+client.gc(older_than=timedelta(days=7))
+client.gc(max_size_bytes=1024 ** 3)  # evict until under 1 GB
+client.invalidate(tags={"experiment": "v1"})
+client.clear()
 
-# Storage stats
 stats = client.stats()
-# {
-#     'total_commits': 42,
-#     'completed_commits': 40,
-#     'stored_objects': 38,      # blob_objects + inline_objects
-#     'disk_bytes': 10485760,    # blob_bytes + inline_bytes
-#     'blob_objects': 35,
-#     'blob_bytes': 9437184,
-#     'inline_objects': 3,
-#     'inline_bytes': 1048576,
-# }
+# {'total_commits', 'completed_commits', 'stored_objects', 'disk_bytes',
+#  'blob_objects', 'blob_bytes', 'inline_objects', 'inline_bytes'}
 ```
 
-### `client.export(path)` / `client.import_archive(path)`
+### Export and import
 
-Export the entire cache to a portable `.tar.gz` archive and import it elsewhere. Blobs are content-addressable, so deduplication is preserved across stores.
+Export the whole cache to a portable archive and import it elsewhere. Blobs are content-addressed, so deduplication is preserved across stores. Use this for migrations (SQLite to Redis), CI cache warm-up, or backups.
 
 ```python
-# Export
 client.export("backup.tar.gz")
 
-# Import into a fresh store
 client2 = Client(store_dir=".cashet2")
-count = client2.import_archive("backup.tar.gz")
-print(f"Imported {count} commits")
+result = client2.import_archive("backup.tar.gz")
+print(f"imported {result.imported}, skipped {result.skipped}")
 ```
 
-Use this for migrations (SQLite → Redis), CI cache warm-up, or backups. Existing commits are skipped during import.
+`import_archive` verifies every blob's content hash and returns an `ImportResult(imported, skipped)`. Commits whose blobs are missing from the archive are skipped and reported rather than silently dropped; existing commits are skipped without re-import.
 
-### Jupyter & Notebook Support
-
-`cashet` works seamlessly in Jupyter notebooks, IPython, and the Python REPL. It uses a tiered source-resolution strategy:
-
-1. **`inspect.getsource()`** — for normal `.py` files
-2. **`dill.source.getsource()`** — for interactive sessions with live history
-3. **`dis.Bytecode` fallback** — for any live function, even after a kernel restart
-
-This means you can define functions in a notebook cell, rerun the cell with changes, and `cashet` will correctly invalidate the cache based on the new code.
+### HTTP server endpoints
 
 ```python
-# In a notebook cell
-client = Client()
-
-def preprocess(data):
-    return [x * 2 for x in data]
+from cashet import Client
 
-ref = client.submit(preprocess, [1, 2, 3])
+client = Client()
+client.serve(host="127.0.0.1", port=8000, require_token="secret123")
 ```
 
-Change the cell body and rerun — the cache invalidates automatically.
+| Method | Path | Description |
+|---|---|---|
+| POST | `/submit` | Run a registered task |
+| GET | `/result/{hash}` | Fetch the deserialized result |
+| GET | `/commit/{hash}` | Commit metadata |
+| GET | `/log` | List commits (`?func=`, `?limit=`, `?status=`) |
+| GET | `/stats` | Storage statistics |
+| POST | `/gc` | Run garbage collection |
 
-### Thread Safety
+When `require_token` is set, every request needs an `Authorization: Bearer <token>` header. Request bodies are size-limited (default 500 MB). Use `AsyncClient.serve()` for the native async server.
 
-`cashet` is safe to use from multiple threads and processes sharing the same store directory. Concurrent submissions of the same uncached task are deduplicated: the function executes **exactly once** and all callers receive the same cached result. This works across `multiprocessing.Process`, `ProcessPoolExecutor`, and multiple independent Python interpreters.
+> **Security.** `/submit` does not execute client-supplied Python by default. The legacy `func_source`, `func_b64`, `args_b64`, and `kwargs_b64` payloads require both `allow_remote_code=True` and a non-empty token (`cashet serve --require-token secret123 --allow-remote-code`). That mode deserializes and runs arbitrary Python, so expose it only to trusted clients.
 
-> **Note:** Cross-process dedup uses a 5-minute timeout by default. If a process dies while running a task, its claim is automatically reclaimed after that timeout so other workers are not blocked forever. You can adjust this via `LocalExecutor(running_ttl=...)`:
->
-> ```python
-> from datetime import timedelta
-> from cashet.executor import LocalExecutor
->
-> client = Client(executor=LocalExecutor(running_ttl=timedelta(minutes=10)))
-> ```
+### Serialization
 
 ```python
-import threading
-
-def worker():
-    c = Client()  # separate Client instance, same store
-    c.submit(expensive_func, arg)
-
-threads = [threading.Thread(target=worker) for _ in range(10)]
-for t in threads:
-    t.start()
-for t in threads:
-    t.join()
-# expensive_func ran only once
-```
-
-### `ResultRef`
-
-A lazy reference to a stored result. Pass it as an argument to chain tasks:
+from cashet import Client, PickleSerializer, SafePickleSerializer, JsonSerializer
 
-```python
-step1 = client.submit(func_a, input_data)
-step2 = client.submit(func_b, step1)  # step1 auto-resolves to its output
-step3 = client.submit(func_c, {"payload": step2})  # nested refs resolve too
+Client(serializer=PickleSerializer())                       # default, arbitrary Python objects
+Client(serializer=SafePickleSerializer())                   # restrict unpickling to an allowlist
+Client(serializer=SafePickleSerializer(extra_classes=[MyClass]))
+Client(serializer=JsonSerializer())                         # JSON-safe data
 ```
 
-`ResultRef` is generic — `submit()` infers the return type from the function annotation:
+Implement the `Serializer` protocol (`dumps`/`loads`) for MessagePack or any custom format.
 
-```python
-ref: ResultRef[int] = client.submit(double, 5)
-result = ref.load()  # typed as int
-```
+### Pluggable backends
 
-### Custom Serialization
+Everything is protocol-based, so you can swap the store, executor, or serializer without changing task code.
 
 ```python
-from cashet import Client, PickleSerializer, SafePickleSerializer, JsonSerializer
-
-# Default: pickle (handles arbitrary Python objects)
-client = Client(serializer=PickleSerializer())
-
-# Safe pickle: restricts deserialization to an allowlist of known types
-client = Client(serializer=SafePickleSerializer())
-
-# Allow custom classes through the allowlist
-client = Client(serializer=SafePickleSerializer(extra_classes=[MyClass]))
-
-# For JSON-safe data (dicts, lists, primitives)
-client = Client(serializer=JsonSerializer())
-
-# Or implement the Serializer protocol
-from cashet.hashing import Serializer
+from cashet import Client
+from cashet.store import SQLiteStore
+from cashet.executor import LocalExecutor
 
-class MySerializer:
-    def dumps(self, obj) -> bytes:
-        ...
-    def loads(self, data: bytes):
-        ...
+client = Client(store=SQLiteStore(".cashet"), executor=LocalExecutor())
 ```
 
+| Protocol | Default | Built-in alternatives | Implement for |
+|---|---|---|---|
+| `Store` | `SQLiteStore` | `RedisStore` | RocksDB, S3, Postgres |
+| `AsyncStore` | `AsyncSQLiteStore` | `AsyncRedisStore` | async variants |
+| `Executor` | `LocalExecutor` | | Celery, Kafka, RQ |
+| `AsyncExecutor` | `AsyncLocalExecutor` | | Celery, Kafka, RQ |
+| `Serializer` | `PickleSerializer` | `JsonSerializer`, `SafePickleSerializer` | MessagePack, custom |
+
 ## How It Works
 
 ```
 client.submit(func, arg1, arg2)
          │
          ▼
-  ┌─────────────────┐
-  │  Hash function   │  SHA256(AST-normalized source + dep versions + referenced user helpers)
+  ┌──────────────────┐
+  │  Hash function   │  SHA256(AST-normalized source + dep versions + referenced helpers)
   │  Hash arguments  │  SHA256(canonical repr of args/kwargs)
-  └────────┬────────┘
+  └────────┬─────────┘
            │
            ▼
-  ┌─────────────────┐
+  ┌──────────────────┐
   │  Fingerprint     │  func_hash:args_hash
   │  cache lookup    │  ← Store protocol (SQLiteStore, RedisStore, ...)
-  └────────┬────────┘
+  └────────┬─────────┘
            │
      ┌─────┴─────┐
-     │            │
-  CACHED       MISS
+   CACHED       MISS
      │            │
      ▼            ▼
-  Return ref   ← Executor protocol (LocalExecutor, CeleryExecutor, ...)
-               Execute function
-               Store result as blob → Store protocol
-               Record commit with parent lineage
-               Return ref
+  Return ref   Execute (Executor protocol), store result as a blob,
+               record a commit with parent lineage, return ref
 ```
 
-**Architecture (protocol-based):**
-
-| Protocol | Default | Built-in alternatives | Implement for |
-|---|---|---|---|
-| `Store` | `SQLiteStore` | `RedisStore` | RocksDB, S3, Postgres |
-| `AsyncStore` | `AsyncSQLiteStore` | `AsyncRedisStore` | async variants of above |
-| `Executor` | `LocalExecutor` | — | Celery, Kafka, RQ |
-| `AsyncExecutor` | `AsyncLocalExecutor` | — | Celery, Kafka, RQ |
-| `Serializer` | `PickleSerializer` | `JsonSerializer`, `SafePickleSerializer` | MessagePack, custom |
-
-**Storage layout** (in `.cashet/`):
+Storage layout under `.cashet/`:
 
 ```
 .cashet/
-├── objects/          # content-addressable blobs (like git objects)
-│   ├── a3/
-│   │   └── b4c5d6... # compressed result blob
-│   └── e7/
-│       └── f8g9h0...
-└── meta.db           # SQLite: commits, fingerprints, provenance, inline_objects
+├── objects/          # content-addressable blobs, like git objects
+│   └── a3/b4c5d6...   # compressed result blob
+└── meta.db           # SQLite: commits, fingerprints, provenance, inline objects
 ```
 
-**Small objects** (<1KB) are stored inline in `meta.db` instead of the filesystem. This reduces inode overhead for caches with many tiny results. Larger objects are stored as compressed blobs in `objects/` as usual.
+Small results (under 1 KB) are stored inline in `meta.db` to avoid inode overhead; larger results are compressed blobs in `objects/`.
 
-**Key design decisions:**
+### Key design decisions
 
-- **Closure variables are not hashed** and emit a `ClosureWarning` if present. Function identity is source code, defaults, keyword defaults, immutable referenced globals, and referenced helper functions; not arbitrary runtime state. If you need cache invalidation based on a mutable value, pass it as an explicit argument.
-- **Referenced user-defined helper functions are hashed recursively.** If your cached function calls or references a helper from your own project (via `co_names` / `globals`), that helper's source is included in the cache key. Change the helper and the caller's cache invalidates. Builtin and stdlib functions are skipped. This behavior is automatic and invisible — no decorators or imports needed.
-- **Nested refs resolve through containers.** `ResultRef` and `AsyncResultRef` values inside lists, tuples, sets, frozensets, and dicts are loaded before execution and recorded as commit input refs.
+- **Function identity is source code, not runtime state.** The hash covers AST-normalized source, default and keyword-default values, immutable referenced globals, and referenced user-defined helper functions. Change a helper and the caller's cache invalidates. Builtins and stdlib are skipped.
+- **Closure variables are not hashed** and emit a `ClosureWarning`. To invalidate on a mutable value, pass it as an explicit argument.
+- **Arguments are hashed by value, including objects.** Custom objects are hashed by their `__dict__` and `__slots__` state plus their class module and qualname, so same-named classes from different modules do not collide.
+- **Source is canonicalized with `ast.unparse`,** so comments, docstrings, and whitespace do not invalidate the cache, and hashes stay stable across Python versions.
 - **Blobs are deduplicated by content hash.** Identical results share one blob on disk.
-- **Source is hashed as an AST.** Comments, docstrings, and whitespace changes don't invalidate the cache.
-- **Custom object arguments include their class module and qualname** in the argument hash so same-named classes from different modules do not collide.
-- **Non-cached tasks get unique commit hashes** (timestamp salt) so they always re-execute but still record lineage.
-- **Parent tracking:** Each commit records the hash of the previous commit for the same function+args, forming a history chain you can traverse.
+- **Nested refs resolve through containers.** `ResultRef` values inside lists, tuples, sets, frozensets, and dicts are loaded before execution and recorded as commit inputs.
+- **Non-cached tasks get a timestamp-salted commit hash,** so they always re-execute while still recording lineage.
+
+### Concurrency
+
+cashet is safe across threads, processes, and machines that share one store. Concurrent submissions of the same uncached task are deduplicated: the function runs exactly once and all callers get the same result. Cross-process claims use file locks (SQLite) or per-fingerprint Redis locks, with a heartbeat lease so a crashed worker's claim is reclaimed (default 5 minutes, configurable via `LocalExecutor(running_ttl=...)`).
+
+### Notebooks
+
+cashet resolves function source through `inspect.getsource`, then `dill.source.getsource` for interactive sessions, then a bytecode fallback that survives a kernel restart. Edit a notebook cell, rerun it, and the cache invalidates on the new code.
 
 ## Configuration
 
-- **`CASHET_DIR`** — override the default `.cashet` store directory. Equivalent to passing `store_dir=`.
-- **`CASHET_LOG`** — set to `DEBUG`, `INFO`, `WARNING`, or `ERROR` to print log output to stderr with a `[cashet]` tag. Messages include task fingerprint, function name, commit hash, and duration inline — no structured logging backend needed.
+- **`CASHET_DIR`** overrides the default `.cashet` store directory, equivalent to `store_dir=`.
+- **`CASHET_LOG`** set to `DEBUG`, `INFO`, `WARNING`, or `ERROR` prints logs to stderr with a `[cashet]` tag, including fingerprint, function name, commit hash, and duration.
 
-## Project Status
+## Version Compatibility
+
+Upgrading across a hash-format change does not corrupt anything; old entries simply miss and recompute on first access. To start clean, run `cashet clear` or point at a fresh store directory.
 
-**Beta.** The core (hashing, DAG resolution, fingerprint dedup) is stable. Works reliably for single-machine, multiprocess, and multi-machine (Redis) workflows.
+- **0.4.4 to 0.4.5:** Hashing fixes (slotted-object state, referenced global containers, `ast.unparse` canonicalization) change function and argument cache keys, so results cached by earlier versions recompute on first access. The Redis tag index key scheme changed and is not migrated; rewrite affected commits to rebuild tag indexes. `import_archive` now returns `ImportResult(imported, skipped)` instead of a bare count.
+- **0.3.x to 0.4.0:** Added per-entry TTL and tag-based invalidation. SQLite auto-migrates; old caches stay readable.
+- **0.3.0 to 0.3.1:** Redis blob keys renamed to `cashet:blob:data:{hash}` and stats backfilled once. Clear Redis caches before upgrading if you rely on long-lived reuse.
+- **0.2.x to 0.3.x:** Hash format unified; caches from 0.2.x do not hit on 0.3.x.
+
+## Project Status
 
-Built-in: `SQLiteStore` + `AsyncSQLiteStore`, `RedisStore` + `AsyncRedisStore`, `LocalExecutor` + `AsyncLocalExecutor`, `PickleSerializer` + `JsonSerializer` + `SafePickleSerializer`, HTTP server (`client.serve()`), CLI.
+Beta. The core (hashing, DAG resolution, fingerprint dedup) is stable and works for single-machine, multiprocess, and multi-machine (Redis) workflows.
 
-Not yet built: RocksDB, S3 stores; Celery/Kafka executors. PRs welcome.
+Built in: `SQLiteStore` and `AsyncSQLiteStore`, `RedisStore` and `AsyncRedisStore`, `LocalExecutor` and `AsyncLocalExecutor`, `PickleSerializer`, `JsonSerializer`, and `SafePickleSerializer`, the HTTP server, and the CLI. Not yet built: RocksDB and S3 stores, Celery and Kafka executors. Pull requests welcome.
 
 ## License
 

From 582d7fe622a1f81c3f6446b7fc2035f07cd89f84 Mon Sep 17 00:00:00 2001
From: jolovicdev <dusan.jolovic@proton.me>
Date: Sun, 21 Jun 2026 18:39:10 +0200
Subject: [PATCH 18/18] fix: reject unauthenticated requests before buffering
 the request body

The size-limit middleware runs outside the per-route token check, so its
body-buffering loop could buffer up to max_content_length for an
unauthenticated client before the route returned 401, widening the
memory-DoS surface on token-protected servers.

When a token is configured, the middleware now verifies the Authorization
header and returns 401 before reading any body. The per-route check stays as
defense in depth. The shared _token_authorized helper is reused by both.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md         |  4 +++-
 src/cashet/server.py | 31 +++++++++++++++++++++++++------
 tests/test_server.py | 26 ++++++++++++++++++++++++++
 3 files changed, 54 insertions(+), 7 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c02a463..ea11a8d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -26,7 +26,9 @@
 - Make the Redis blob delete script idempotent against a missing reference
   counter so it cannot drop a blob another commit still references.
 - Enforce the HTTP server request size limit on the bytes actually received,
-  closing a bypass for chunked requests that omit `Content-Length`.
+  closing a bypass for chunked requests that omit `Content-Length`. On
+  token-protected servers, unauthenticated requests are now rejected before any
+  body is buffered.
 - Validate handler input (returning 400) and wrap every server handler in a
   generic 500 barrier so malformed input and internal errors no longer leak
   stack traces; `/gc` also accepts an empty body and falls back to defaults.
diff --git a/src/cashet/server.py b/src/cashet/server.py
index a86e4ce..dec1f9c 100644
--- a/src/cashet/server.py
+++ b/src/cashet/server.py
@@ -200,13 +200,17 @@ def _decode_call(
     return tuple(args), kwargs, None
 
 
+def _token_authorized(auth: str, token: str) -> bool:
+    return auth.startswith("Bearer ") and hmac.compare_digest(auth[7:], token)
+
+
 def _require_token(handler: Any, token: str | None) -> Any:
     if token is None:
         return handler
 
     async def wrapper(request: Request) -> JSONResponse:
         auth = request.headers.get("authorization", "")
-        if not auth.startswith("Bearer ") or not hmac.compare_digest(auth[7:], token):
+        if not _token_authorized(auth, token):
             logger.warning(
                 "request unauthorized method=%s path=%s",
                 request.method,
@@ -416,6 +420,7 @@ def create_async_app(
     app.state.tasks = _server_tasks(client, tasks)
     app.state.allow_remote_code = allow_remote_code
     app.state.max_content_length = max_content_length
+    app.state.require_token = require_token
     return app
 
 
@@ -430,11 +435,24 @@ async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
             await self.app(scope, receive, send)
             return
         starlette_app = scope.get("app")
-        max_size = (
-            getattr(starlette_app.state, "max_content_length", _DEFAULT_MAX_CONTENT_LENGTH)
-            if starlette_app is not None
-            else _DEFAULT_MAX_CONTENT_LENGTH
-        )
+        state = starlette_app.state if starlette_app is not None else None
+        max_size = getattr(state, "max_content_length", _DEFAULT_MAX_CONTENT_LENGTH)
+
+        # Reject unauthenticated requests before buffering any body, so an
+        # unauthenticated client cannot force buffering up to max_content_length.
+        require_token = getattr(state, "require_token", None)
+        if require_token is not None:
+            auth = ""
+            for name, value in scope["headers"]:
+                if name == b"authorization":
+                    auth = value.decode("latin-1")
+                    break
+            if not _token_authorized(auth, require_token):
+                await _CustomJSONResponse(
+                    {"error": "unauthorized"}, status_code=401
+                )(scope, receive, send)
+                return
+
         for name, value in scope["headers"]:
             if name == b"content-length":
                 try:
@@ -687,4 +705,5 @@ def create_app(
     app.state.tasks = _server_tasks(client, tasks)
     app.state.allow_remote_code = allow_remote_code
     app.state.max_content_length = max_content_length
+    app.state.require_token = require_token
     return app
diff --git a/tests/test_server.py b/tests/test_server.py
index 53230c1..61d825f 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -74,6 +74,32 @@ def test_request_within_limit_ok(self, tmp_path: Path) -> None:
         response = tc.post("/submit", json={"task": "add", "args": [3, 4], "kwargs": {}})
         assert response.status_code == 200
 
+    def test_unauthenticated_rejected_before_buffering(self, tmp_path: Path) -> None:
+        client = Client(store_dir=tmp_path / ".cashet")
+        app = create_app(
+            client, tasks={"add": _add}, require_token="secret", max_content_length=200
+        )
+        tc = TestClient(app)
+
+        def body_chunks() -> Iterator[bytes]:
+            for _ in range(10):
+                yield b"x" * 100  # oversized, no Content-Length
+
+        # No token: rejected with 401 before the body is buffered (not 413).
+        response = tc.post("/submit", content=body_chunks())
+        assert response.status_code == 401
+
+    def test_authenticated_request_still_processed(self, tmp_path: Path) -> None:
+        client = Client(store_dir=tmp_path / ".cashet")
+        app = create_app(client, tasks={"add": _add}, require_token="secret")
+        tc = TestClient(app)
+        response = tc.post(
+            "/submit",
+            json={"task": "add", "args": [3, 4]},
+            headers={"Authorization": "Bearer secret"},
+        )
+        assert response.status_code == 200
+
 
 class TestServerValidation:
     def test_log_bad_limit_returns_400(self, server_client: TestClient) -> None: