diff --git a/src/claude_agent_sdk/client.py b/src/claude_agent_sdk/client.py
index 03705f08..19f0d42f 100644
--- a/src/claude_agent_sdk/client.py
+++ b/src/claude_agent_sdk/client.py
@@ -309,9 +309,13 @@ async def query(
         else:
             # Handle AsyncIterable prompts - stream them
             async for msg in prompt:
-                # Ensure session_id is set on each message
+                # Ensure session_id is set on each message without mutating the
+                # caller's dict. Mutating it would make a reused message
+                # iterable ignore a changed session_id on a later query() call
+                # (the "session_id" key would already be present from the first
+                # call), silently sending the stale id.
                 if "session_id" not in msg:
-                    msg["session_id"] = session_id
+                    msg = {**msg, "session_id": session_id}
                 await self._transport.write(json.dumps(msg) + "\n")
 
     async def interrupt(self) -> None:
diff --git a/tests/test_streaming_client.py b/tests/test_streaming_client.py
index 00fe785c..bea3295b 100644
--- a/tests/test_streaming_client.py
+++ b/tests/test_streaming_client.py
@@ -1097,6 +1097,46 @@ def mock_build_command(self):
             # Clean up
             Path(test_script).unlink()
 
+    @pytest.mark.anyio
+    async def test_query_does_not_mutate_caller_messages(self):
+        """query() must not inject session_id into the caller's message dicts.
+
+        Mutating them would make a reused message iterable send a stale
+        session_id on the next query() call, silently ignoring the new one.
+        """
+
+        with patch(
+            "claude_agent_sdk._internal.transport.subprocess_cli.SubprocessCLITransport"
+        ) as mock_transport_class:
+            mock_transport = create_mock_transport()
+            mock_transport_class.return_value = mock_transport
+
+            # Caller-owned, reusable message templates (no session_id set).
+            messages = [{"type": "user", "message": {"role": "user", "content": "hi"}}]
+
+            async def stream():
+                for msg in messages:
+                    yield msg
+
+            async with ClaudeSDKClient() as client:
+                await client.query(stream(), session_id="session-A")
+                # The caller's dict must be untouched.
+                assert "session_id" not in messages[0]
+
+                await client.query(stream(), session_id="session-B")
+
+            sent_session_ids = []
+            for call in mock_transport.write.call_args_list:
+                try:
+                    sent = json.loads(call[0][0].strip())
+                except (json.JSONDecodeError, AttributeError):
+                    continue
+                if sent.get("type") == "user":
+                    sent_session_ids.append(sent.get("session_id"))
+
+            # Each call honors its own session_id argument on reuse.
+            assert sent_session_ids == ["session-A", "session-B"]
+
 
 class TestClaudeSDKClientEdgeCases:
     """Test edge cases and error scenarios."""