release 0.1.2: batch DAGs, notebook support, thread safety, and cache observability

Features:
- Add client.submit_many() with TaskRef for batch submission and
  automatic topological ordering of dependencies.
- Expose ResultRef.commit_hash so users can inspect lineage, diff,
  and manage commits directly from a result reference.
- Add Client.clear() as a convenience wrapper for gc(timedelta(days=0)).
- Add CASHET_DIR environment variable fallback for store_dir.
- CLI improvements: cashet get pretty-prints strings/dicts/lists,
  cashet clear alias, and human-readable disk sizes in stats.

Notebook & dynamic source support:
- Add dill dependency for source extraction in notebooks/REPLs.
- Tiered fallback in get_func_source(): inspect.getsource -> dill ->
  stable bytecode representation (co_code + co_consts).
- Functions defined in Jupyter, IPython, exec(), and lambdas now
  hash correctly and invalidate cache on semantic changes.

Thread safety & correctness:
- Fix LocalExecutor check-then-act race with a shared lock registry
  keyed by store path. Concurrent submits deduplicate across threads
  and even across separate Client instances sharing the same store.
- Fix client.get() to populate commit_hash on the temporary ResultRef.

Robustness:
- Replace string-building _stable_repr with progressive _stable_hash
  to avoid multi-megabyte intermediate strings for large args.
- Add length-prefixed hashing tags to prevent collision attacks.
- Add cycle detection in _stable_repr_to and _stable_hash for
  recursive data structures (lists, dicts, sets, objects).
- Add input validation to submit_many for clear TypeError messages
  on bad task shapes.

Tests:
- Add coverage for submit_many, TaskRef wiring, progressive hash
  collision resistance, disk_bytes stats, CASHET_DIR env var,
  recursive structures, dynamic/bytecode hashing, and concurrent
  deduplication both within one Client and across multiple Clients.
- Verify all 8 pipeline scenarios execute correctly inside an actual
  Jupyter kernel via nbconvert.

Docs:
- Document commit_hash, Client.clear(), Jupyter/REPL support,
  and thread safety guarantees in README.md.

Author: jolovicdev

README.md

@@ -176,7 +176,7 @@ Fix the `join_crm` function and re-run the script. Steps 1-2 return instantly fr
 
 ### 3. Reproducible Notebook Results
 
-Share a result with a colleague and they can verify exactly how it was produced:
+`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
@@ -340,13 +340,22 @@ Submit a function for execution. Returns a `ResultRef` — a lazy handle to the
 
 ```python
 ref = client.submit(my_func, arg1, arg2, key="value")
-ref.hash        # content hash of the result
-ref.size        # size in bytes
-ref.load()      # deserialize and return the result
+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
 ```
 
 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))`.
+
+```python
+client.clear()
+```
+
 ### `client.submit_many(tasks) -> list[ResultRef]`
 
 Submit a batch of tasks with automatic topological ordering. Use `TaskRef(index)` to wire outputs between tasks in the batch.
@@ -445,6 +454,47 @@ stats = client.stats()
 # {'total_commits': 42, 'completed_commits': 40, 'stored_objects': 38, 'disk_bytes': 10485760}
 ```
 
+### 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.
+
+```python
+# In a notebook cell
+client = Client()
+
+def preprocess(data):
+    return [x * 2 for x in data]
+
+ref = client.submit(preprocess, [1, 2, 3])
+```
+
+Change the cell body and rerun — the cache invalidates automatically.
+
+### Thread Safety
+
+`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.
+
+```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:

pyproject.toml

@@ -21,6 +21,7 @@ classifiers = [
 requires-python = ">=3.11"
 dependencies = [
     "click>=8.1",
+    "dill>=0.3",
     "rich>=13.0",
 ]
 

src/cashet/client.py

@@ -92,7 +92,7 @@ def submit(
         )
         if commit.output_ref is None:
             raise RuntimeError(f"Task {commit.task_def.func_name} failed: {commit.error}")
-        ref = ResultRef(commit.output_ref, self.store, self.serializer)
+        ref = ResultRef(commit.output_ref, self.store, self.serializer, commit_hash=commit.hash)
         return ref
 
     @overload
@@ -168,7 +168,9 @@ def get(self, hash: str) -> Any:
             raise KeyError(f"No commit found with hash {hash}")
         if commit.output_ref is None:
             raise ValueError(f"Commit {hash[:12]} has no output")
-        ref = ResultRef(commit.output_ref, self.store, self.serializer)
+        ref = ResultRef(
+            commit.output_ref, self.store, self.serializer, commit_hash=commit.hash
+        )
         return ref.load()
 
     def diff(self, hash_a: str, hash_b: str) -> dict[str, Any]:
@@ -189,6 +191,9 @@ def gc(self, older_than: timedelta | None = None) -> int:
         cutoff = datetime.now(UTC) - ttl
         return self.store.evict(cutoff)
 
+    def clear(self) -> int:
+        return self.gc(timedelta(days=0))
+
     def close(self) -> None:
         self.store.close()
 
@@ -342,7 +347,7 @@ def _execute_batch(
         )
         if commit.output_ref is None:
             raise RuntimeError(f"Task {commit.task_def.func_name} failed: {commit.error}")
-        ref = ResultRef(commit.output_ref, store, serializer)
+        ref = ResultRef(commit.output_ref, store, serializer, commit_hash=commit.hash)
         results[key] = ref
     return results
 

src/cashet/dag.py

@@ -26,12 +26,19 @@ def resolve_input_refs(args: tuple[Any, ...], kwargs: dict[str, Any]) -> list[Ob
 
 
 class ResultRef:
-    __slots__ = ("_loaded", "_ref", "_serializer", "_store", "_value")
-
-    def __init__(self, ref: ObjectRef, store: Store, serializer: Serializer) -> None:
+    __slots__ = ("_commit_hash", "_loaded", "_ref", "_serializer", "_store", "_value")
+
+    def __init__(
+        self,
+        ref: ObjectRef,
+        store: Store,
+        serializer: Serializer,
+        commit_hash: str = "",
+    ) -> None:
         self._ref = ref
         self._store = store
         self._serializer = serializer
+        self._commit_hash = commit_hash
         self._value: Any = None
         self._loaded = False
 
@@ -42,6 +49,10 @@ def __cashet_ref__(self) -> ObjectRef:
     def hash(self) -> str:
         return self._ref.hash
 
+    @property
+    def commit_hash(self) -> str:
+        return self._commit_hash
+
     @property
     def short_hash(self) -> str:
         return self._ref.short()
@@ -58,7 +69,8 @@ def load(self) -> Any:
         return self._value
 
     def __repr__(self) -> str:
-        return f"ResultRef(hash={self.short_hash}, size={self.size}, loaded={self._loaded})"
+        ch = self._commit_hash[:12] if self._commit_hash else "?"
+        return f"ResultRef(commit={ch}, blob={self.short_hash}, size={self.size})"
 
 
 def compute_commit_hash(

src/cashet/executor.py

@@ -2,6 +2,7 @@
 
 import threading
 import traceback
+from pathlib import Path
 from typing import Any
 
 from cashet.dag import (
@@ -14,11 +15,26 @@
 from cashet.models import Commit, ObjectRef, TaskDef, TaskStatus
 from cashet.protocols import Store
 
+_STORE_LOCKS: dict[str, threading.Lock] = {}
+_STORE_LOCKS_GUARD = threading.Lock()
+_FALLBACK_LOCK = threading.Lock()
 
-class LocalExecutor:
-    def __init__(self) -> None:
-        self._lock = threading.Lock()
 
+def _get_store_lock(store: Store) -> threading.Lock:
+    root = getattr(store, "root", None)
+    if root is not None:
+        key = str(Path(root).resolve())
+        with _STORE_LOCKS_GUARD:
+            if key not in _STORE_LOCKS:
+                _STORE_LOCKS[key] = threading.Lock()
+            return _STORE_LOCKS[key]
+    store_lock = getattr(store, "_lock", None)
+    if store_lock is not None:
+        return store_lock
+    return _FALLBACK_LOCK
+
+
+class LocalExecutor:
     def submit(
         self,
         func: Any,
@@ -28,7 +44,7 @@ def submit(
         store: Store,
         serializer: Serializer,
     ) -> tuple[Commit, bool]:
-        with self._lock:
+        with _get_store_lock(store):
             input_refs = resolve_input_refs(args, kwargs)
             existing = find_existing_commit(store, task_def)
             if existing is not None: