Home / Function/ test_thread_count_excludes_blocked_run_calls() — langchain Function Reference

test_thread_count_excludes_blocked_run_calls() — langchain Function Reference

Architecture documentation for the test_thread_count_excludes_blocked_run_calls() function in test_tool_call_limit.py from the langchain codebase.

Entity Profile

Dependency Diagram

graph TD
  60dc1af6_02e1_bd50_21ac_5db10d7bb7e2["test_thread_count_excludes_blocked_run_calls()"]
  a75b8390_08d3_7137_c8a7_9d78fc0c4517["test_tool_call_limit.py"]
  60dc1af6_02e1_bd50_21ac_5db10d7bb7e2 -->|defined in| a75b8390_08d3_7137_c8a7_9d78fc0c4517
  style 60dc1af6_02e1_bd50_21ac_5db10d7bb7e2 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_call_limit.py lines 489–532

def test_thread_count_excludes_blocked_run_calls() -> None:
    """Test that thread count only includes allowed calls, not blocked run-scoped calls.

    When run_limit is lower than thread_limit and multiple parallel calls are made,
    only the allowed calls should increment the thread count.

    Example: If run_limit=1 and 3 parallel calls are made, thread count should be 1
    (not 3) because the other 2 were blocked by the run limit.
    """
    # Set run_limit=1, thread_limit=10 (much higher)
    middleware = ToolCallLimitMiddleware(thread_limit=10, run_limit=1, exit_behavior="continue")
    runtime = None

    # Make 3 parallel tool calls - only 1 should be allowed by run_limit
    state = ToolCallLimitState(
        messages=[
            AIMessage(
                "Response",
                tool_calls=[
                    {"name": "search", "args": {}, "id": "1"},
                    {"name": "search", "args": {}, "id": "2"},
                    {"name": "search", "args": {}, "id": "3"},
                ],
            )
        ],
        thread_tool_call_count={},
        run_tool_call_count={},
    )
    result = middleware.after_model(state, runtime)  # type: ignore[arg-type]
    assert result is not None

    # Thread count should be 1 (only the allowed call)
    assert result["thread_tool_call_count"]["__all__"] == 1, (
        "Thread count should only include the 1 allowed call, not the 2 blocked calls"
    )
    # Run count should be 3 (all attempted calls)
    assert result["run_tool_call_count"]["__all__"] == 3, (
        "Run count should include all 3 attempted calls"
    )

    # Verify 2 error messages were created for blocked calls
    assert "messages" in result
    error_messages = [msg for msg in result["messages"] if isinstance(msg, ToolMessage)]
    assert len(error_messages) == 2, "Should have 2 error messages for the 2 blocked calls"

Domain

Subdomains

Frequently Asked Questions

What does test_thread_count_excludes_blocked_run_calls() do?
test_thread_count_excludes_blocked_run_calls() is a function in the langchain codebase, defined in libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_call_limit.py.
Where is test_thread_count_excludes_blocked_run_calls() defined?
test_thread_count_excludes_blocked_run_calls() is defined in libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_call_limit.py at line 489.

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free