diff --git a/tests/integration/fixtures/text_command.yaml b/tests/integration/fixtures/text_command.yaml
new file mode 100644
index 0000000000..cc91e2f792
--- /dev/null
+++ b/tests/integration/fixtures/text_command.yaml
@@ -0,0 +1,37 @@
+esphome:
+  name: host-text-command-test
+
+host:
+
+api:
+  batch_delay: 0ms
+
+logger:
+
+text:
+  - platform: template
+    name: "Test Text"
+    id: test_text
+    optimistic: true
+    min_length: 0
+    max_length: 255
+    mode: text
+    initial_value: "initial"
+
+  - platform: template
+    name: "Test Password"
+    id: test_password
+    optimistic: true
+    min_length: 4
+    max_length: 32
+    mode: password
+    initial_value: "secret"
+
+  - platform: template
+    name: "Test Text Long"
+    id: test_text_long
+    optimistic: true
+    min_length: 0
+    max_length: 255
+    mode: text
+    initial_value: ""
diff --git a/tests/integration/test_text_command.py b/tests/integration/test_text_command.py
new file mode 100644
index 0000000000..82fe981578
--- /dev/null
+++ b/tests/integration/test_text_command.py
@@ -0,0 +1,126 @@
+"""Integration test for text command zero-copy optimization.
+
+Tests that TextCommandRequest correctly handles the pointer_to_buffer
+optimization for the state field, ensuring text values are properly
+transmitted via the API.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from aioesphomeapi import TextInfo, TextState
+import pytest
+
+from .state_utils import InitialStateHelper, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_text_command(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test text command with various string values including edge cases."""
+    loop = asyncio.get_running_loop()
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        # Verify we can get device info
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "host-text-command-test"
+
+        # Get list of entities
+        entities, _ = await client.list_entities_services()
+
+        # Find our text entities using require_entity
+        test_text = require_entity(entities, "test_text", TextInfo, "Test Text entity")
+        test_password = require_entity(
+            entities, "test_password", TextInfo, "Test Password entity"
+        )
+        test_text_long = require_entity(
+            entities, "test_text_long", TextInfo, "Test Text Long entity"
+        )
+
+        # Track state changes
+        states: dict[int, Any] = {}
+        state_futures: dict[int, asyncio.Future[Any]] = {}
+
+        def on_state(state: Any) -> None:
+            states[state.key] = state
+            if state.key in state_futures and not state_futures[state.key].done():
+                state_futures[state.key].set_result(state)
+
+        # Set up InitialStateHelper to swallow initial state broadcasts
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        # Wait for all initial states to be received
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Verify initial states were received
+        assert test_text.key in initial_state_helper.initial_states
+        initial_text_state = initial_state_helper.initial_states[test_text.key]
+        assert isinstance(initial_text_state, TextState)
+        assert initial_text_state.state == "initial"
+
+        async def wait_for_state_change(key: int, timeout: float = 2.0) -> Any:
+            """Wait for a state change for the given entity key."""
+            state_futures[key] = loop.create_future()
+            try:
+                return await asyncio.wait_for(state_futures[key], timeout)
+            finally:
+                state_futures.pop(key, None)
+
+        # Test 1: Simple text value
+        client.text_command(key=test_text.key, state="hello world")
+        state = await wait_for_state_change(test_text.key)
+        assert state.state == "hello world"
+
+        # Test 2: Empty string (edge case for zero-copy)
+        client.text_command(key=test_text.key, state="")
+        state = await wait_for_state_change(test_text.key)
+        assert state.state == ""
+
+        # Test 3: Single character
+        client.text_command(key=test_text.key, state="x")
+        state = await wait_for_state_change(test_text.key)
+        assert state.state == "x"
+
+        # Test 4: String with special characters
+        client.text_command(key=test_text.key, state="hello\tworld\n!")
+        state = await wait_for_state_change(test_text.key)
+        assert state.state == "hello\tworld\n!"
+
+        # Test 5: Unicode characters
+        client.text_command(key=test_text.key, state="hello 世界 🌍")
+        state = await wait_for_state_change(test_text.key)
+        assert state.state == "hello 世界 🌍"
+
+        # Test 6: Long string (tests buffer handling)
+        long_text = "a" * 200
+        client.text_command(key=test_text_long.key, state=long_text)
+        state = await wait_for_state_change(test_text_long.key)
+        assert state.state == long_text
+        assert len(state.state) == 200
+
+        # Test 7: Password field (same mechanism, different mode)
+        client.text_command(key=test_password.key, state="newpassword123")
+        state = await wait_for_state_change(test_password.key)
+        assert state.state == "newpassword123"
+
+        # Test 8: String with null bytes embedded (edge case)
+        # Note: protobuf strings should handle this but it's good to verify
+        client.text_command(key=test_text.key, state="before\x00after")
+        state = await wait_for_state_change(test_text.key)
+        assert state.state == "before\x00after"
+
+        # Test 9: Rapid successive commands (tests buffer reuse)
+        for i in range(5):
+            client.text_command(key=test_text.key, state=f"rapid_{i}")
+            state = await wait_for_state_change(test_text.key)
+            assert state.state == f"rapid_{i}"