fix(checkpoint): make TTL operations best-effort to prevent write loss (#174)

* fix(checkpoint): make all TTL operations best-effort to prevent write loss

EXPIRE commands mixed into critical write pipelines caused complete
pipeline failure on Redis Enterprise proxy when TTL commands were routed
to different shards than the JSON module commands. This lost interrupt
writes silently, breaking human-in-the-loop workflows.

Changes:
- Remove EXPIRE from all put_writes pipelines (aio, sync, shallow)
- Use raise_on_error=False for pipeline execution with per-result
  inspection to preserve JSON.MERGE fallback logic
- Apply TTL individually per key after pipeline success, wrapped in
  try/except so failures only log warnings
- Wrap all remaining bare EXPIRE/PERSIST calls across the codebase
  (put, get_tuple, key_registry, _apply_ttl_to_keys) in try/except
- Simplify _apply_ttl_to_keys to use individual calls instead of
  building EXPIRE-only pipelines that share the same failure mode

* test(checkpoint): add HitL+TTL regression tests and update mocked TTL tests

- Add test_hitl_ttl_regression.py with 9 tests covering:
  - HitL+TTL integration guards on real Redis (5 tests)
  - Pipeline no longer contains EXPIRE commands
  - Writes survive simulated TTL failure
  - TTL failure logs warning via caplog
  - TTL applied separately after pipeline succeeds
- Update test_base_client_info_and_ttl.py to expect individual expire
  calls instead of pipeline-based TTL
- Update test_cluster_mode.py to remove cluster/non-cluster branching
  since both modes now use individual expire calls

* fix(checkpoint): address PR review comments on TTL warning diagnostics

- Add exc_info=True and key names to all TTL warning logs for
  diagnosability (aio.py, shallow.py, ashallow.py, key_registry.py)
- Remove redundant outer try/except around _key_registry.apply_ttl()
  in __init__.py since it's already best-effort internally
- Remove dead cluster_mode variable in base.py _apply_ttl_to_keys
- Remove unused imports in test_hitl_ttl_regression.py
- Remove xfail marker from test_sync_hitl_with_ttl (was masking a
  local environment issue, not a real bug)

* fix(serializer): support both old and new Interrupt field layouts

The Interrupt dataclass changed between langgraph versions:
- langgraph <=1.0.x: Interrupt(value, id)
- langgraph >=1.1.x: Interrupt(value, resumable, ns, when)

The serializer now introspects the installed Interrupt class at runtime
via dataclasses.fields() and serializes/deserializes whichever fields
are present. This ensures checkpoints written by either version can be
read by either version without data loss.

* fix(test): use state.next for sync interrupt detection

Sync get_state() in langgraph 1.0.x does not populate
task.interrupts (only aget_state does). Use state.next to
verify the graph is paused at the interrupt node instead.

Also reorder the sync test before async tests to avoid any
potential event loop contamination.

Author: Brian Sam-Bodden

langgraph/checkpoint/redis/__init__.py

@@ -581,9 +581,15 @@ def put(
         )
         self._redis.set(latest_pointer_key, checkpoint_key)
 
-        # Apply TTL to latest pointer key as well
+        # Apply TTL to latest pointer key as well (best-effort)
         if ttl_seconds is not None:
-            self._redis.expire(latest_pointer_key, ttl_seconds)
+            try:
+                self._redis.expire(latest_pointer_key, ttl_seconds)
+            except Exception:
+                logger.warning(
+                    "Failed to apply TTL to latest pointer key: %s",
+                    latest_pointer_key,
+                )
 
         return next_config
 
@@ -618,11 +624,16 @@ def put_writes(
             }
             writes_objects.append(write_obj)
 
-        with self._redis.pipeline(transaction=False) as pipeline:
-            # Keep track of keys we're creating
-            created_keys = []
-            write_keys = []
+        # IMPORTANT: Only critical commands (JSON.SET, JSON.MERGE) go in the pipeline.
+        # EXPIRE (TTL) commands are applied separately afterward to avoid pipeline
+        # failures on Redis Enterprise proxy, where mixed JSON module + native commands
+        # in a single pipeline can cause EXPIRE to fail, aborting the entire pipeline
+        # and losing interrupt writes.
+        write_keys: list[str] = []
+        checkpoint_key = ""
+        merge_failed = False
 
+        with self._redis.pipeline(transaction=False) as pipeline:
             for write_obj in writes_objects:
                 idx_value = write_obj["idx"]
                 assert isinstance(idx_value, int)
@@ -635,80 +646,58 @@ def put_writes(
                 )
                 write_keys.append(key)
                 pipeline.json().set(key, "$", cast(Any, write_obj))
-                created_keys.append(key)
 
-            # Add TTL operations to the pipeline if configured
-            if created_keys and self.ttl_config and "default_ttl" in self.ttl_config:
-                ttl_seconds = int(self.ttl_config["default_ttl"] * 60)
-                for key in created_keys:
-                    pipeline.expire(key, ttl_seconds)
-
-            # Update checkpoint to indicate it has writes
+            # Update checkpoint to indicate it has writes (critical)
             if writes_objects:
                 checkpoint_key = self._make_redis_checkpoint_key_cached(
                     thread_id, checkpoint_ns, checkpoint_id
                 )
-                # Use merge to update existing document
                 pipeline.json().merge(checkpoint_key, "$", {"has_writes": True})
 
+            # Execute critical commands with raise_on_error=False
+            results = pipeline.execute(raise_on_error=False)
+
+            # Check results for critical command failures
+            for result in results:
+                if isinstance(result, Exception):
+                    err_str = str(result)
+                    if "JSON.MERGE" in err_str or "merge" in err_str.lower():
+                        merge_failed = True
+                    else:
+                        raise result
+
+        # Handle JSON.MERGE fallback for older Redis versions
+        if merge_failed and checkpoint_key:
             try:
-                pipeline.execute()
-            except Exception as e:
-                # Check if JSON.MERGE failed (older Redis versions)
-                if "JSON.MERGE" in str(e) or "merge" in str(e).lower():
-                    # Retry without JSON.MERGE for older Redis versions
-                    with self._redis.pipeline(transaction=False) as fallback_pipeline:
-                        # Re-add all the write operations
-                        for write_obj in writes_objects:
-                            idx_value = write_obj["idx"]
-                            assert isinstance(idx_value, int)
-                            key = self._make_redis_checkpoint_writes_key_cached(
-                                thread_id,
-                                checkpoint_ns,
-                                checkpoint_id,
-                                task_id,
-                                idx_value,
-                            )
-                            fallback_pipeline.json().set(key, "$", cast(Any, write_obj))
-
-                        # Add TTL operations if configured
-                        if (
-                            created_keys
-                            and self.ttl_config
-                            and "default_ttl" in self.ttl_config
-                        ):
-                            ttl_seconds = int(self.ttl_config["default_ttl"] * 60)
-                            for key in created_keys:
-                                fallback_pipeline.expire(key, ttl_seconds)
-
-                        # Execute the fallback pipeline
-                        fallback_pipeline.execute()
-
-                        # Update has_writes flag separately for older Redis
-                        if checkpoint_key:
-                            try:
-                                checkpoint_data = self._redis.json().get(checkpoint_key)
-                                if isinstance(
-                                    checkpoint_data, dict
-                                ) and not checkpoint_data.get("has_writes"):
-                                    checkpoint_data["has_writes"] = True
-                                    self._redis.json().set(
-                                        checkpoint_key, "$", checkpoint_data
-                                    )
-                            except Exception:
-                                # If this fails, it's not critical - the writes are still saved
-                                pass
-                else:
-                    # Re-raise other exceptions
-                    raise
+                checkpoint_data = self._redis.json().get(checkpoint_key)
+                if isinstance(checkpoint_data, dict) and not checkpoint_data.get(
+                    "has_writes"
+                ):
+                    checkpoint_data["has_writes"] = True
+                    self._redis.json().set(checkpoint_key, "$", checkpoint_data)
+            except Exception:
+                pass
+
+        # Apply TTL separately (best-effort — failures here don't lose writes).
+        # Individual calls ensure partial success: if one key's EXPIRE fails
+        # on RE proxy, the others still get TTL applied.
+        if write_keys and self.ttl_config and "default_ttl" in self.ttl_config:
+            ttl_seconds = int(self.ttl_config["default_ttl"] * 60)
+            for key in write_keys:
+                try:
+                    self._redis.expire(key, ttl_seconds)
+                except Exception:
+                    logger.warning(
+                        "Failed to apply TTL to checkpoint write key: %s", key
+                    )
 
         # Update key registry with the write keys
         if self._key_registry and write_keys:
             self._key_registry.register_write_keys_batch(
                 thread_id, checkpoint_ns, checkpoint_id, write_keys
             )
 
-            # Apply TTL to registry key if configured
+            # Apply TTL to registry key (already best-effort inside apply_ttl)
             if self.ttl_config and "default_ttl" in self.ttl_config:
                 ttl_seconds = int(self.ttl_config["default_ttl"] * 60)
                 self._key_registry.apply_ttl(