fix: enable fine-grained tool streaming when include_partial_messages=True (#644)

## Problem

When `include_partial_messages=True`, the SDK passes
`--include-partial-messages` to the CLI, which tells the CLI to emit
`stream_event` messages. However, tool input parameters are still
**buffered by the API** and never streamed — `input_json_delta` events
are withheld until the entire tool input is assembled.

**Root cause:** CLI v2.1.40 (shipped Feb 10 2026) removed the always-on
`fine-grained-tool-streaming-2025-05-14` beta header and replaced it
with two opt-in paths:
1. A GrowthBook feature flag (`tengu_fgts`, default: `false`)
2. The `CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING` env var → sets
`eager_input_streaming: true` per-tool in the API request

The SDK was never updated to set the env var, so
`include_partial_messages=True` silently stopped delivering tool input
deltas for users without the GrowthBook flag.

**Regression window:** **v0.1.36** (released Feb 13 2026, bundled CLI
2.1.42) through present. Last working version: **v0.1.35** (bundled CLI
2.1.39).

Fixes #608. Same fix applied to the TypeScript SDK in parallel.

## Fix

When `include_partial_messages=True`, inject
`CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING=1` into the subprocess
environment via `setdefault` (user-supplied values take precedence).

## Tests

Three new unit tests in `test_transport.py`:
- `include_partial_messages=True` → env var set to `"1"`
- `include_partial_messages=False` → env var not set
- User-supplied value in `options.env` is respected

```
$ python -m pytest tests/test_transport.py -v -k fgts

tests/test_transport.py::TestSubprocessCLITransport::test_include_partial_messages_enables_fgts PASSED
tests/test_transport.py::TestSubprocessCLITransport::test_include_partial_messages_false_does_not_set_fgts PASSED
tests/test_transport.py::TestSubprocessCLITransport::test_user_can_override_fgts_env_var PASSED

3 passed in 0.35s
```

## Live E2E verification

Ran against the bundled CLI (v2.1.49, post-regression) with fix applied
— confirmed `input_json_delta` events stream through the wire:

```
$ env -u CLAUDECODE PYTHONPATH=src python e2e_fgts.py

Running E2E FGTS test...
Prompt: "Use the Bash tool to run: echo 'Fine-grained tool streaming test'"

  input_json_delta [  1]: ''
  input_json_delta [  2]: '{"command":'
  input_json_delta [  3]: ' "echo'
  input_json_delta [  4]: ' \'Fine-gra'
  input_json_delta [  5]: 'ined tool st'
  input_json_delta [  6]: 'reamin'
  input_json_delta [  7]: 'g tes'
  input_json_delta [  8]: 't\'"}'

Total stream_event messages:  23
input_json_delta events:      8
Assembled tool input JSON:    '{"command": "echo \'Fine-grained tool streaming test\'"}'
PASS: tool input deltas streamed correctly
```

**Note:** The "before" (broken) path cannot be demonstrated on this
account because the `tengu_fgts` GrowthBook flag is currently on for
authenticated users, which is the second condition in the CLI's FGTS
check:
```typescript
if (getFeatureValue_CACHED_MAY_BE_STALE('tengu_fgts', false) || // on for us
    isEnvTruthy(process.env.CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING)) {
  schema.eager_input_streaming = true
}
```
The fix matters for users where the GrowthBook flag is off (the default
is `false`).

Author: qing-ant

src/claude_agent_sdk/_internal/transport/subprocess_cli.py

@@ -354,6 +354,15 @@ async def connect(self) -> None:
             if self._options.enable_file_checkpointing:
                 process_env["CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING"] = "true"
 
+            # Enable fine-grained tool streaming when partial messages are requested.
+            # --include-partial-messages emits stream_event messages, but tool input
+            # parameters are still buffered by the API unless eager_input_streaming is
+            # also enabled at the per-tool level via this env var.
+            if self._options.include_partial_messages:
+                process_env.setdefault(
+                    "CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING", "1"
+                )
+
             if self._cwd:
                 process_env["PWD"] = self._cwd
 

tests/test_transport.py

@@ -883,6 +883,138 @@ def test_build_command_always_uses_streaming(self):
         assert "stream-json" in cmd
         assert "--print" not in cmd
 
+    def test_include_partial_messages_enables_fgts(self):
+        """Test that include_partial_messages=True sets CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING.
+
+        --include-partial-messages tells the CLI to forward stream_event messages,
+        but tool input parameters are still buffered by the API unless
+        eager_input_streaming is enabled via this env var.
+        """
+
+        async def _test():
+            options = make_options(include_partial_messages=True)
+
+            with patch(
+                "anyio.open_process", new_callable=AsyncMock
+            ) as mock_open_process:
+                mock_version_process = MagicMock()
+                mock_version_process.stdout = MagicMock()
+                mock_version_process.stdout.receive = AsyncMock(
+                    return_value=b"2.0.0 (Claude Code)"
+                )
+                mock_version_process.terminate = MagicMock()
+                mock_version_process.wait = AsyncMock()
+
+                mock_process = MagicMock()
+                mock_process.stdout = MagicMock()
+                mock_stdin = MagicMock()
+                mock_stdin.aclose = AsyncMock()
+                mock_process.stdin = mock_stdin
+                mock_process.returncode = None
+
+                mock_open_process.side_effect = [mock_version_process, mock_process]
+
+                transport = SubprocessCLITransport(
+                    prompt="test",
+                    options=options,
+                )
+                await transport.connect()
+
+                second_call_kwargs = mock_open_process.call_args_list[1].kwargs
+                env_passed = second_call_kwargs["env"]
+                assert (
+                    env_passed.get("CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING")
+                    == "1"
+                )
+
+        anyio.run(_test)
+
+    def test_include_partial_messages_false_does_not_set_fgts(self):
+        """Test that include_partial_messages=False does not force-enable FGTS."""
+
+        async def _test():
+            options = make_options(include_partial_messages=False)
+
+            with patch(
+                "anyio.open_process", new_callable=AsyncMock
+            ) as mock_open_process:
+                mock_version_process = MagicMock()
+                mock_version_process.stdout = MagicMock()
+                mock_version_process.stdout.receive = AsyncMock(
+                    return_value=b"2.0.0 (Claude Code)"
+                )
+                mock_version_process.terminate = MagicMock()
+                mock_version_process.wait = AsyncMock()
+
+                mock_process = MagicMock()
+                mock_process.stdout = MagicMock()
+                mock_stdin = MagicMock()
+                mock_stdin.aclose = AsyncMock()
+                mock_process.stdin = mock_stdin
+                mock_process.returncode = None
+
+                mock_open_process.side_effect = [mock_version_process, mock_process]
+
+                transport = SubprocessCLITransport(
+                    prompt="test",
+                    options=options,
+                )
+                await transport.connect()
+
+                second_call_kwargs = mock_open_process.call_args_list[1].kwargs
+                env_passed = second_call_kwargs["env"]
+                # Should not be set (unless the user already had it in their env)
+                assert (
+                    "CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING" not in env_passed
+                )
+
+        anyio.run(_test)
+
+    def test_user_can_override_fgts_env_var(self):
+        """Test that a user-supplied env var takes precedence over the SDK default."""
+
+        async def _test():
+            options = make_options(
+                include_partial_messages=True,
+                env={"CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING": "0"},
+            )
+
+            with patch(
+                "anyio.open_process", new_callable=AsyncMock
+            ) as mock_open_process:
+                mock_version_process = MagicMock()
+                mock_version_process.stdout = MagicMock()
+                mock_version_process.stdout.receive = AsyncMock(
+                    return_value=b"2.0.0 (Claude Code)"
+                )
+                mock_version_process.terminate = MagicMock()
+                mock_version_process.wait = AsyncMock()
+
+                mock_process = MagicMock()
+                mock_process.stdout = MagicMock()
+                mock_stdin = MagicMock()
+                mock_stdin.aclose = AsyncMock()
+                mock_process.stdin = mock_stdin
+                mock_process.returncode = None
+
+                mock_open_process.side_effect = [mock_version_process, mock_process]
+
+                transport = SubprocessCLITransport(
+                    prompt="test",
+                    options=options,
+                )
+                await transport.connect()
+
+                second_call_kwargs = mock_open_process.call_args_list[1].kwargs
+                env_passed = second_call_kwargs["env"]
+                # User's explicit "0" should win over SDK default "1"
+                assert (
+                    env_passed.get("CLAUDE_CODE_ENABLE_FINE_GRAINED_TOOL_STREAMING")
+                    == "0"
+                )
+
+        anyio.run(_test)
+
     def test_build_command_large_agents_work(self):
         """Test that large agent definitions work without size limits.