sdk-py 0.2.9

- When connection is dropped while waiting, reconnect up to 5 times if a
Location header is present

libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

@@ -450,7 +450,7 @@ class PostgresSaver(BasePostgresSaver):
             {
                 **value["checkpoint"],
                 "channel_values": {
-                    **value["checkpoint"].get("channel_values"),
+                    **(value["checkpoint"].get("channel_values") or {}),
                     **self._load_blobs(value["channel_values"]),
                 },
             },

libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

@@ -409,7 +409,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
             {
                 **value["checkpoint"],
                 "channel_values": {
-                    **value["checkpoint"].get("channel_values"),
+                    **(value["checkpoint"].get("channel_values") or {}),
                     **self._load_blobs(value["channel_values"]),
                 },
             },

libs/checkpoint-postgres/tests/test_async.py

@@ -344,3 +344,34 @@ async def test_pending_sends_migration(saver_name: str) -> None:
             TASKS: ["send-1", "send-2", "send-3"]
         }
         assert TASKS in search_results[0].checkpoint["channel_versions"]
+
+
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+async def test_get_checkpoint_no_channel_values(
+    monkeypatch, saver_name: str, test_data
+) -> None:
+    """Backwards compatibility test that verifies a checkpoint with no channel_values key can be retrieved without throwing an error."""
+    async with _saver(saver_name) as saver:
+        config = {
+            "configurable": {
+                "thread_id": "thread-2",
+                "checkpoint_ns": "",
+                "__super_private_key": "super_private_value",
+            },
+            "metadata": {"run_id": "my_run_id"},
+        }
+        chkpnt: Checkpoint = create_checkpoint(empty_checkpoint(), {}, 1)
+        await saver.aput(config, chkpnt, {}, {})
+
+        load_checkpoint_tuple = saver._load_checkpoint_tuple
+
+        def patched_load_checkpoint_tuple(value):
+            value["checkpoint"].pop("channel_values", None)
+            return load_checkpoint_tuple(value)
+
+        monkeypatch.setattr(
+            saver, "_load_checkpoint_tuple", patched_load_checkpoint_tuple
+        )
+
+        checkpoint = await saver.aget_tuple(config)
+        assert checkpoint.checkpoint["channel_values"] == {}

libs/checkpoint-postgres/tests/test_sync.py

@@ -332,3 +332,33 @@ def test_pending_sends_migration(saver_name: str) -> None:
             TASKS: ["send-1", "send-2", "send-3"]
         }
         assert TASKS in search_results[0].checkpoint["channel_versions"]
+
+
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+def test_get_checkpoint_no_channel_values(
+    monkeypatch, saver_name: str, test_data
+) -> None:
+    """Backwards compatibility test that verifies a checkpoint with no channel_values key can be retrieved without throwing an error."""
+    with _saver(saver_name) as saver:
+        config = {
+            "configurable": {
+                "thread_id": "thread-2",
+                "checkpoint_ns": "",
+                "__super_private_key": "super_private_value",
+            },
+        }
+        chkpnt: Checkpoint = create_checkpoint(empty_checkpoint(), {}, 1)
+        saver.put(config, chkpnt, {}, {})
+
+        load_checkpoint_tuple = saver._load_checkpoint_tuple
+
+        def patched_load_checkpoint_tuple(value):
+            value["checkpoint"].pop("channel_values", None)
+            return load_checkpoint_tuple(value)
+
+        monkeypatch.setattr(
+            saver, "_load_checkpoint_tuple", patched_load_checkpoint_tuple
+        )
+
+        checkpoint = saver.get_tuple(config)
+        assert checkpoint.checkpoint["channel_values"] == {}

libs/checkpoint-sqlite/tests/test_sqlite.py

@@ -116,7 +116,17 @@ class TestSqliteSaver:
                 search_results_5[1].config["configurable"]["checkpoint_ns"],
             } == {"", "inner"}
 
-            # TODO: test before and limit params
+            # search with before param
+            search_results_6 = list(saver.list(None, before=search_results_5[1].config))
+            assert len(search_results_6) == 1
+            assert search_results_6[0].config["configurable"]["thread_id"] == "thread-1"
+
+            # search with limit param
+            search_results_7 = list(
+                saver.list({"configurable": {"thread_id": "thread-2"}}, limit=1)
+            )
+            assert len(search_results_7) == 1
+            assert search_results_7[0].config["configurable"]["thread_id"] == "thread-2"
 
     def test_search_where(self) -> None:
         # call method / assertions

libs/cli/tests/unit_tests/test_util.py

@@ -0,0 +1,153 @@
+from unittest.mock import patch
+
+from langgraph_cli.util import clean_empty_lines, warn_non_wolfi_distro
+
+
+def test_clean_empty_lines():
+    """Test clean_empty_lines function."""
+    # Test with empty lines
+    input_str = "line1\n\nline2\n\nline3"
+    result = clean_empty_lines(input_str)
+    assert result == "line1\nline2\nline3"
+    
+    # Test with no empty lines
+    input_str = "line1\nline2\nline3"
+    result = clean_empty_lines(input_str)
+    assert result == "line1\nline2\nline3"
+    
+    # Test with only empty lines
+    input_str = "\n\n\n"
+    result = clean_empty_lines(input_str)
+    assert result == ""
+    
+    # Test empty string
+    input_str = ""
+    result = clean_empty_lines(input_str)
+    assert result == ""
+
+
+def test_warn_non_wolfi_distro_with_debian(capsys):
+    """Test that warning is shown when image_distro is 'debian'."""
+    config = {"image_distro": "debian"}
+    
+    warn_non_wolfi_distro(config)
+    
+    captured = capsys.readouterr()
+    assert "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security." in captured.out
+    assert "Wolfi is a security-oriented, minimal Linux distribution designed for containers." in captured.out
+    assert 'To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.' in captured.out
+
+
+def test_warn_non_wolfi_distro_with_default_debian(capsys):
+    """Test that warning is shown when image_distro is missing (defaults to debian)."""
+    config = {}  # No image_distro key, should default to debian
+    
+    warn_non_wolfi_distro(config)
+    
+    captured = capsys.readouterr()
+    assert "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security." in captured.out
+    assert "Wolfi is a security-oriented, minimal Linux distribution designed for containers." in captured.out
+    assert 'To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.' in captured.out
+
+
+def test_warn_non_wolfi_distro_with_wolfi(capsys):
+    """Test that no warning is shown when image_distro is 'wolfi'."""
+    config = {"image_distro": "wolfi"}
+    
+    warn_non_wolfi_distro(config)
+    
+    captured = capsys.readouterr()
+    assert captured.out == ""  # No output should be generated
+
+
+def test_warn_non_wolfi_distro_with_other_distro(capsys):
+    """Test that warning is shown when image_distro is something other than 'wolfi'."""
+    config = {"image_distro": "ubuntu"}
+    
+    warn_non_wolfi_distro(config)
+    
+    captured = capsys.readouterr()
+    assert "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security." in captured.out
+    assert "Wolfi is a security-oriented, minimal Linux distribution designed for containers." in captured.out
+    assert 'To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.' in captured.out
+
+
+def test_warn_non_wolfi_distro_output_formatting():
+    """Test that the warning output is properly formatted with colors and empty line."""
+    config = {"image_distro": "debian"}
+    
+    with patch('click.secho') as mock_secho:
+        warn_non_wolfi_distro(config)
+    
+    # Verify click.secho was called with the correct parameters
+    expected_calls = [
+        (
+            ("⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security.",),
+            {"fg": "yellow", "bold": True}
+        ),
+        (
+            ("   Wolfi is a security-oriented, minimal Linux distribution designed for containers.",),
+            {"fg": "yellow"}
+        ),
+        (
+            ('   To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.',),
+            {"fg": "yellow"}
+        ),
+        (
+            ("",),  # Empty line
+            {}
+        )
+    ]
+    
+    assert mock_secho.call_count == 4
+    for i, (expected_args, expected_kwargs) in enumerate(expected_calls):
+        actual_call = mock_secho.call_args_list[i]
+        assert actual_call.args == expected_args
+        assert actual_call.kwargs == expected_kwargs
+
+
+def test_warn_non_wolfi_distro_various_configs(capsys):
+    """Test warn_non_wolfi_distro with various config scenarios."""
+    test_cases = [
+        # (config, should_warn, description)
+        ({"image_distro": "debian"}, True, "explicit debian"),
+        ({"image_distro": "wolfi"}, False, "explicit wolfi"),
+        ({}, True, "missing image_distro (defaults to debian)"),
+        ({"image_distro": "alpine"}, True, "other distro"),
+        ({"image_distro": "ubuntu"}, True, "ubuntu distro"),
+        ({"other_config": "value"}, True, "unrelated config keys"),
+    ]
+    
+    for config, should_warn, description in test_cases:
+        # Clear any previous output
+        capsys.readouterr()
+        
+        warn_non_wolfi_distro(config)
+        
+        captured = capsys.readouterr()
+        if should_warn:
+            assert "⚠️  Security Recommendation" in captured.out, f"Should warn for {description}"
+            assert "Wolfi" in captured.out, f"Should mention Wolfi for {description}"
+        else:
+            assert captured.out == "", f"Should not warn for {description}"
+
+
+def test_warn_non_wolfi_distro_return_value():
+    """Test that warn_non_wolfi_distro returns None."""
+    config = {"image_distro": "debian"}
+    result = warn_non_wolfi_distro(config)
+    assert result is None
+    
+    config = {"image_distro": "wolfi"}
+    result = warn_non_wolfi_distro(config)
+    assert result is None
+
+
+def test_warn_non_wolfi_distro_does_not_modify_config():
+    """Test that warn_non_wolfi_distro does not modify the input config."""
+    original_config = {"image_distro": "debian", "other_key": "value"}
+    config_copy = original_config.copy()
+    
+    warn_non_wolfi_distro(config_copy)
+    
+    assert config_copy == original_config  # Config should remain unchanged

libs/langgraph/langgraph/pregel/_loop.py

@@ -242,7 +242,9 @@ class PregelLoop:
         self.interrupt_before = interrupt_before
         self.manager = manager
         self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
-        self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
+        self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF] or (
+            CONFIG_KEY_RESUMING in self.config[CONF] and self.is_nested
+        )
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy

libs/langgraph/tests/test_pregel.py

@@ -3445,6 +3445,73 @@ def test_stream_buffering_single_node(sync_checkpointer: BaseCheckpointSaver) ->
     ]
 
 
+def test_nested_graph_resume_reuses_cached_task_writes(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    # Reproduces issue where a helper @task inside a nested graph re-executes
+    # on resume instead of reusing cached writes. Ensures it runs only once.
+    counter_parent = 0
+    counter_sub = 0
+
+    @task
+    def get_time_parent() -> float:
+        nonlocal counter_parent
+        counter_parent += 1
+        return time.time()
+
+    @task
+    def get_time_subgraph() -> float:
+        nonlocal counter_sub
+        counter_sub += 1
+        return time.time()
+
+    class State(TypedDict):
+        state_counter: int
+
+    # Subgraph that calls a helper task and then interrupts
+    sub = StateGraph(State)
+
+    def human_node(_: State):
+        _ = get_time_subgraph().result()
+        interrupt("what is your name?")
+
+    sub.add_node("human_node", human_node)
+    sub.set_entry_point("human_node")
+    sub.set_finish_point("human_node")
+    subgraph = sub.compile(checkpointer=sync_checkpointer)
+
+    # Parent graph that calls a helper task and interrupts, then enters subgraph
+    parent = StateGraph(State)
+
+    def parent_node(_: State):
+        _ = get_time_parent().result()
+        interrupt("what is your parent name?")
+
+    parent.add_node("parent_node", parent_node)
+    parent.add_node("subgraph", subgraph)
+    parent.add_edge(START, "parent_node")
+    parent.add_edge("parent_node", "subgraph")
+    parent.add_edge("subgraph", END)
+    graph = parent.compile(checkpointer=sync_checkpointer)
+
+    cfg_parent = {"configurable": {"thread_id": str(uuid.uuid4())}}
+
+    # First run – interrupts in parent node
+    for _ in graph.stream({"state_counter": 1}, cfg_parent):
+        pass
+
+    # Resume 1 – proceeds into subgraph, interrupts there
+    for _ in graph.stream(Command(resume="resume-1"), cfg_parent):
+        pass
+
+    # Resume 2 – completes without re-running subgraph helper task
+    for _ in graph.stream(Command(resume="resume-2"), cfg_parent):
+        pass
+
+    assert counter_parent == 1
+    assert counter_sub == 1
+
+
 def test_nested_graph_interrupts_parallel(
     sync_checkpointer: BaseCheckpointSaver, durability: Durability
 ) -> None:

libs/sdk-py/langgraph_sdk/__init__.py

@@ -1,6 +1,6 @@
 from langgraph_sdk.auth import Auth
 from langgraph_sdk.client import get_client, get_sync_client
 
-__version__ = "0.2.8"
+__version__ = "0.2.9"
 
 __all__ = ["Auth", "get_client", "get_sync_client"]

libs/sdk-py/langgraph_sdk/auth/__init__.py

@@ -728,7 +728,7 @@ def is_studio_user(
     return (
         isinstance(user, types.StudioUser)
         or isinstance(user, dict)
-        and user.get("kind") == "StudioUser"
+        and user.get("kind") == "StudioUser"  # ty: ignore[invalid-argument-type]
     )
 
 

libs/sdk-py/langgraph_sdk/client.py

@@ -437,6 +437,54 @@ class HttpClient:
                 logger.error(f"Error from langgraph-api: {body}", exc_info=e)
             raise e
 
+    async def request_reconnect(
+        self,
+        path: str,
+        method: str,
+        *,
+        json: dict[str, Any] | None = None,
+        params: QueryParamTypes | None = None,
+        headers: Mapping[str, str] | None = None,
+        on_response: Callable[[httpx.Response], None] | None = None,
+        reconnect_limit: int = 5,
+    ) -> Any:
+        """Send a request that automatically reconnects to Location header."""
+        request_headers, content = await _aencode_json(json)
+        if headers:
+            request_headers.update(headers)
+        async with self.client.stream(
+            method, path, headers=request_headers, content=content, params=params
+        ) as r:
+            if on_response:
+                on_response(r)
+            try:
+                r.raise_for_status()
+            except httpx.HTTPStatusError as e:
+                body = (await r.aread()).decode()
+                if sys.version_info >= (3, 11):
+                    e.add_note(body)
+                else:
+                    logger.error(f"Error from langgraph-api: {body}", exc_info=e)
+                raise e
+            loc = r.headers.get("location")
+            if reconnect_limit <= 0 or not loc:
+                return await _adecode_json(r)
+            try:
+                return await _adecode_json(r)
+            except httpx.HTTPError:
+                warnings.warn(
+                    f"Request failed, attempting reconnect to Location: {loc}",
+                    stacklevel=2,
+                )
+                await r.aclose()
+                return await self.request_reconnect(
+                    loc,
+                    "GET",
+                    headers=request_headers,
+                    # don't pass on_response so it's only called once
+                    reconnect_limit=reconnect_limit - 1,
+                )
+
     async def stream(
         self,
         path: str,
@@ -2533,8 +2581,9 @@ class RunsClient:
             if on_run_created and (metadata := _get_run_metadata_from_response(res)):
                 on_run_created(metadata)
 
-        response = await self.http.post(
+        response = await self.http.request_reconnect(
             endpoint,
+            "POST",
             json={k: v for k, v in payload.items() if v is not None},
             params=params,
             headers=headers,
@@ -2679,12 +2728,20 @@ class RunsClient:
         }
         if params:
             query_params.update(params)
-        return await self.http.post(
-            f"/threads/{thread_id}/runs/{run_id}/cancel",
-            json=None,
-            params=query_params,
-            headers=headers,
-        )
+        if wait:
+            return await self.http.request_reconnect(
+                f"/threads/{thread_id}/runs/{run_id}/cancel",
+                "POST",
+                params=query_params,
+                headers=headers,
+            )
+        else:
+            return await self.http.post(
+                f"/threads/{thread_id}/runs/{run_id}/cancel",
+                json=None,
+                params=query_params,
+                headers=headers,
+            )
 
     async def join(
         self,
@@ -2716,8 +2773,11 @@ class RunsClient:
             ```
 
         """  # noqa: E501
-        return await self.http.get(
-            f"/threads/{thread_id}/runs/{run_id}/join", headers=headers, params=params
+        return await self.http.request_reconnect(
+            f"/threads/{thread_id}/runs/{run_id}/join",
+            "GET",
+            headers=headers,
+            params=params,
         )
 
     def join_stream(
@@ -3689,6 +3749,54 @@ class SyncHttpClient:
                 logger.error(f"Error from langgraph-api: {body}", exc_info=e)
             raise e
 
+    def request_reconnect(
+        self,
+        path: str,
+        method: str,
+        *,
+        json: dict[str, Any] | None = None,
+        params: QueryParamTypes | None = None,
+        headers: Mapping[str, str] | None = None,
+        on_response: Callable[[httpx.Response], None] | None = None,
+        reconnect_limit: int = 5,
+    ) -> Any:
+        """Send a request that automatically reconnects to Location header."""
+        request_headers, content = _encode_json(json)
+        if headers:
+            request_headers.update(headers)
+        with self.client.stream(
+            method, path, headers=request_headers, content=content, params=params
+        ) as r:
+            if on_response:
+                on_response(r)
+            try:
+                r.raise_for_status()
+            except httpx.HTTPStatusError as e:
+                body = r.read().decode()
+                if sys.version_info >= (3, 11):
+                    e.add_note(body)
+                else:
+                    logger.error(f"Error from langgraph-api: {body}", exc_info=e)
+                raise e
+            loc = r.headers.get("location")
+            if reconnect_limit <= 0 or not loc:
+                return _decode_json(r)
+            try:
+                return _decode_json(r)
+            except httpx.HTTPError:
+                warnings.warn(
+                    f"Request failed, attempting reconnect to Location: {loc}",
+                    stacklevel=2,
+                )
+                r.close()
+                return self.request_reconnect(
+                    loc,
+                    "GET",
+                    headers=request_headers,
+                    # don't pass on_response so it's only called once
+                    reconnect_limit=reconnect_limit - 1,
+                )
+
     def stream(
         self,
         path: str,
@@ -5754,8 +5862,9 @@ class SyncRunsClient:
         endpoint = (
             f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait"
         )
-        return self.http.post(
+        return self.http.request_reconnect(
             endpoint,
+            "POST",
             json={k: v for k, v in payload.items() if v is not None},
             params=params,
             headers=headers,
@@ -5878,11 +5987,25 @@ class SyncRunsClient:
             ```
 
         """  # noqa: E501
+        query_params = {
+            "wait": 1 if wait else 0,
+            "action": action,
+        }
+        if params:
+            query_params.update(params)
+        if wait:
+            return self.http.request_reconnect(
+                f"/threads/{thread_id}/runs/{run_id}/cancel",
+                "POST",
+                json=None,
+                params=query_params,
+                headers=headers,
+            )
         return self.http.post(
-            f"/threads/{thread_id}/runs/{run_id}/cancel?wait={1 if wait else 0}&action={action}",
+            f"/threads/{thread_id}/runs/{run_id}/cancel",
             json=None,
+            params=query_params,
             headers=headers,
-            params=params,
         )
 
     def join(
@@ -5915,8 +6038,11 @@ class SyncRunsClient:
             ```
 
         """  # noqa: E501
-        return self.http.get(
-            f"/threads/{thread_id}/runs/{run_id}/join", headers=headers, params=params
+        return self.http.request_reconnect(
+            f"/threads/{thread_id}/runs/{run_id}/join",
+            "GET",
+            headers=headers,
+            params=params,
         )
 
     def join_stream(