applied the fix + the windows permission issue, you can ignore if you don't want it but ran all of the tests and passed

Author: 20ns

libs/core/langchain_core/messages/utils.py

@@ -1440,11 +1440,10 @@ def _first_max_tokens(
         # When all messages fit, only apply end_on filtering if needed
         if end_on:
             for _ in range(len(messages)):
-                if not _is_message_type(messages[-1], end_on):
-                    messages.pop()
-                else:
+                if not messages or _is_message_type(messages[-1], end_on):
                     break
-        return messages
+                messages.pop()
+        return _remove_orphaned_tool_messages(messages)
 
     # Use binary search to find the maximum number of messages within token limit
     left, right = 0, len(messages)
@@ -1535,7 +1534,7 @@ def _first_max_tokens(
             else:
                 break
 
-    return messages[:idx]
+    return _remove_orphaned_tool_messages(messages[:idx])
 
 
 def _last_max_tokens(
@@ -1594,7 +1593,42 @@ def _last_max_tokens(
     if system_message:
         result = [system_message, *result]
 
-    return result
+    return _remove_orphaned_tool_messages(result)
+
+
+def _remove_orphaned_tool_messages(
+    messages: Sequence[BaseMessage],
+) -> list[BaseMessage]:
+    """Drop tool messages whose corresponding tool calls are absent."""
+    if not messages:
+        return []
+
+    valid_tool_call_ids: set[str] = set()
+    for message in messages:
+        if isinstance(message, AIMessage):
+            if message.tool_calls:
+                for tool_call in message.tool_calls:
+                    tool_call_id = tool_call.get("id")
+                    if tool_call_id:
+                        valid_tool_call_ids.add(tool_call_id)
+            if isinstance(message.content, list):
+                for block in message.content:
+                    if (
+                        isinstance(block, dict)
+                        and block.get("type") == "tool_use"
+                        and block.get("id")
+                    ):
+                        valid_tool_call_ids.add(block["id"])
+
+    cleaned_messages: list[BaseMessage] = []
+    for message in messages:
+        if isinstance(message, ToolMessage) and (
+            not valid_tool_call_ids
+            or message.tool_call_id not in valid_tool_call_ids
+        ):
+            continue
+        cleaned_messages.append(message)
+    return cleaned_messages
 
 
 _MSG_CHUNK_MAP: dict[type[BaseMessage], type[BaseMessageChunk]] = {

libs/core/tests/unit_tests/messages/test_utils.py

@@ -393,6 +393,84 @@ def test_trim_messages_last_30_include_system_allow_partial_end_on_human() -> No
     assert _MESSAGES_TO_TRIM == _MESSAGES_TO_TRIM_COPY
 
 
+def test_trim_messages_last_removes_orphaned_tool_message() -> None:
+    messages = [
+        HumanMessage("What's the weather in Florida?"),
+        AIMessage(
+            [
+                {"type": "text", "text": "Let's check the weather in Florida"},
+                {
+                    "type": "tool_use",
+                    "id": "abc123",
+                    "name": "get_weather",
+                    "input": {"location": "Florida"},
+                },
+            ],
+            tool_calls=[
+                {
+                    "name": "get_weather",
+                    "args": {"location": "Florida"},
+                    "id": "abc123",
+                    "type": "tool_call",
+                }
+            ],
+        ),
+        ToolMessage("It's sunny.", name="get_weather", tool_call_id="abc123"),
+        HumanMessage("I see"),
+        AIMessage("Do you want to know anything else?"),
+        HumanMessage("No, thanks"),
+        AIMessage("You're welcome! Have a great day!"),
+    ]
+
+    trimmed = trim_messages(
+        messages,
+        strategy="last",
+        token_counter=len,
+        max_tokens=5,
+    )
+
+    expected = [
+        HumanMessage("I see"),
+        AIMessage("Do you want to know anything else?"),
+        HumanMessage("No, thanks"),
+        AIMessage("You're welcome! Have a great day!"),
+    ]
+
+    assert trimmed == expected
+    assert _MESSAGES_TO_TRIM == _MESSAGES_TO_TRIM_COPY
+
+
+def test_trim_messages_last_preserves_tool_message_when_call_present() -> None:
+    messages = [
+        HumanMessage("Start"),
+        AIMessage(
+            "Sure, let me check",
+            tool_calls=[
+                {
+                    "name": "search",
+                    "args": {"query": "status"},
+                    "id": "tool-1",
+                    "type": "tool_call",
+                }
+            ],
+        ),
+        ToolMessage("All systems operational", tool_call_id="tool-1"),
+        HumanMessage("Thanks"),
+    ]
+
+    trimmed = trim_messages(
+        messages,
+        strategy="last",
+        token_counter=len,
+        max_tokens=3,
+    )
+
+    expected = messages[1:]
+
+    assert trimmed == expected
+    assert _MESSAGES_TO_TRIM == _MESSAGES_TO_TRIM_COPY
+
+
 def test_trim_messages_last_40_include_system_allow_partial_start_on_human() -> None:
     expected = [
         SystemMessage("This is a 4 token text."),

libs/core/tests/unit_tests/prompts/test_prompt.py

@@ -1,8 +1,9 @@
 """Test functionality related to prompts."""
 
 import re
+from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Any, Literal, Union
+from typing import Any, Literal, Union, cast
 from unittest import mock
 
 import pytest
@@ -18,6 +19,11 @@
 PYDANTIC_VERSION_AT_LEAST_29 = version.parse("2.9") <= PYDANTIC_VERSION
 
 
+def _normalize_blank_lines(text: str) -> str:
+    """Collapse whitespace-only lines to simplify cross-platform comparisons."""
+    return "\n".join("" if not line.strip() else line for line in text.splitlines())
+
+
 def test_prompt_valid() -> None:
     """Test prompts can be constructed."""
     template = "This is a {foo} test."
@@ -33,19 +39,22 @@ def test_from_file_encoding() -> None:
     input_variables = ["foo"]
 
     # First write to a file using CP-1252 encoding.
-    with NamedTemporaryFile(delete=True, mode="w", encoding="cp1252") as f:
+    with NamedTemporaryFile(delete=False, mode="w", encoding="cp1252") as f:
         f.write(template)
         f.flush()
-        file_name = f.name
+        file_path = Path(f.name)
 
+    try:
         # Now read from the file using CP-1252 encoding and test
-        prompt = PromptTemplate.from_file(file_name, encoding="cp1252")
+        prompt = PromptTemplate.from_file(file_path, encoding="cp1252")
         assert prompt.template == template
         assert prompt.input_variables == input_variables
 
         # Now read from the file using UTF-8 encoding and test
         with pytest.raises(UnicodeDecodeError):
-            PromptTemplate.from_file(file_name, encoding="utf-8")
+            PromptTemplate.from_file(file_path, encoding="utf-8")
+    finally:
+        file_path.unlink(missing_ok=True)
 
 
 def test_prompt_from_template() -> None:
@@ -216,10 +225,12 @@ def test_mustache_prompt_from_template(snapshot: SnapshotAssertion) -> None:
         {{bar}}
     {{/foo}}is a test."""
     prompt = PromptTemplate.from_template(template, template_format="mustache")
-    assert prompt.format(foo=[{"bar": "yo"}, {"bar": "hello"}]) == (
+    assert _normalize_blank_lines(
+        prompt.format(foo=[{"bar": "yo"}, {"bar": "hello"}])
+    ) == (
         """This
         yo
-    
+
         hello
     is a test."""  # noqa: W293
     )
@@ -347,7 +358,7 @@ def test_prompt_invalid_template_format() -> None:
         PromptTemplate(
             input_variables=input_variables,
             template=template,
-            template_format="bar",
+            template_format=cast(PromptTemplateFormat, "bar"),
         )
 
 
@@ -681,7 +692,7 @@ def test_prompt_with_template_variable_name_jinja2() -> None:
 
 def test_prompt_template_add_with_with_another_format() -> None:
     with pytest.raises(ValueError, match=r"Cannot add templates"):
-        (
+        _ = (
             PromptTemplate.from_template("This is a {template}")
             + PromptTemplate.from_template("So {{this}} is", template_format="mustache")
         )

libs/core/tests/unit_tests/test_imports.py

@@ -1,6 +1,7 @@
 import concurrent.futures
 import importlib
 import subprocess
+import sys
 from pathlib import Path
 
 
@@ -26,7 +27,7 @@ def try_to_import(module_name: str) -> tuple[int, str]:
         getattr(module, cls_)
 
     result = subprocess.run(
-        ["python", "-c", f"import langchain_core.{module_name}"], check=True
+        [sys.executable, "-c", f"import langchain_core.{module_name}"], check=True
     )
     return result.returncode, module_name